diff --git a/docs/features/reasoning_outputs.md b/docs/features/reasoning_outputs.md index 16f2a675f8..53f0d16245 100644 --- a/docs/features/reasoning_outputs.md +++ b/docs/features/reasoning_outputs.md @@ -29,6 +29,7 @@ vLLM currently supports the following reasoning models: | [MiniMax-M2](https://huggingface.co/MiniMaxAI/MiniMax-M2) | `minimax_m2_append_think` | `json`, `regex` | ✅ | | [Qwen3 series](https://huggingface.co/collections/Qwen/qwen3-67dd247413f0e2e4f653967f) | `qwen3` | `json`, `regex` | ✅ | | [QwQ-32B](https://huggingface.co/Qwen/QwQ-32B) | `deepseek_r1` | `json`, `regex` | ✅ | +| [Step-3.5 / Step-3.7 Flash](https://huggingface.co/stepfun-ai/Step-3.5-Flash) | `step3p5` | `json`, `regex` | ✅ | !!! note IBM Granite 3.2 and DeepSeek-V3.1 reasoning is disabled by default; to enable it, you must also pass `thinking=True` in your `chat_template_kwargs`. diff --git a/docs/features/tool_calling.md b/docs/features/tool_calling.md index 6527eccd09..bbbc4d6bcd 100644 --- a/docs/features/tool_calling.md +++ b/docs/features/tool_calling.md @@ -469,6 +469,18 @@ Use the checkpoint's chat template with `--tool-call-parser mimo --reasoning-par For schema-constrained tool arguments, set `strict: true` inside each tool's `function`, or use `--tool-strict-level parameter`. The parser supports `auto`, `required`, named tools, and `parallel_tool_calls=False`, preserving leading/trailing whitespace in string arguments. +### Step-3.5 and Step-3.7 Models (`step3p5`) + +Supported models: + +* `stepfun-ai/Step-3.5-Flash` +* `stepfun-ai/Step-3.5-Flash-FP8` +* `stepfun-ai/Step-3.7-Flash` + +Flags: `--tool-call-parser step3p5 --reasoning-parser step3p5 --enable-auto-tool-choice` + +The parser supports `auto`, `required`, and named tools. + ### Olmo 3 Models (`olmo3`) Olmo 3 models output tool calls in a format that is very similar to the one expected by the `pythonic` parser (see below), with a few differences. Each tool call is a pythonic string, but the parallel tool calls are newline-delimited, and the calls are wrapped within XML tags as `..`. In addition, the parser also allows JSON boolean and null literals (`true`, `false`, and `null`) in addition to the pythonic ones (`True`, `False`, and `None`). diff --git a/tests/parser/engine/test_step3p5.py b/tests/parser/engine/test_step3p5.py new file mode 100644 index 0000000000..68f4dc55ef --- /dev/null +++ b/tests/parser/engine/test_step3p5.py @@ -0,0 +1,122 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Tests for the engine-based Step-3.5 parser. + +Step-3.5 reuses the Qwen3 grammar, which ``test_qwen3.py`` covers. These tests +cover what differs: thinking is always on, the newline the model emits before +```` is dropped, prior-turn reasoning does not end the current turn, +and the ``step3p5`` names resolve to the engine adapters. +""" + +import json + +import pytest + +from tests.parser.engine.conftest import make_mock_tokenizer +from tests.parser.engine.streaming_helpers import simulate_reasoning_streaming +from vllm.parser.engine.registered_adapters import ( + Step3p5ParserReasoningAdapter, + Step3p5ParserToolAdapter, +) +from vllm.parser.step3p5 import Step3p5Parser +from vllm.reasoning import ReasoningParserManager +from vllm.tool_parsers import ToolParserManager + +_THINK_START_ID = 50 +_THINK_END_ID = 51 +_IM_START_ID = 70 +_IM_END_ID = 71 +_TEXT_ID = 100 + +_STEP3P5_VOCAB = { + "": _THINK_START_ID, + "": _THINK_END_ID, + "": 60, + "": 61, + "<|im_start|>": _IM_START_ID, + "<|im_end|>": _IM_END_ID, +} + + +@pytest.fixture +def mock_tokenizer(): + return make_mock_tokenizer(_STEP3P5_VOCAB) + + +@pytest.fixture +def parser(mock_tokenizer): + return Step3p5Parser(mock_tokenizer) + + +def test_parser_names_resolve_to_engine_adapters(): + tool_parser_cls = ToolParserManager.get_tool_parser("step3p5") + assert issubclass(tool_parser_cls, Step3p5ParserToolAdapter) + assert tool_parser_cls.structural_tag_model == "qwen_3_coder" + reasoning_parser_cls = ReasoningParserManager.get_reasoning_parser("step3p5") + assert reasoning_parser_cls is Step3p5ParserReasoningAdapter + + +def test_enable_thinking_false_is_ignored(mock_tokenizer): + # The chat template always prefills "\n" and has no thinking switch. + parser = Step3p5Parser( + mock_tokenizer, chat_template_kwargs={"enable_thinking": False} + ) + assert parser.extract_reasoning("plananswer", None) == ("plan", "answer") + assert not parser.is_reasoning_end([_IM_START_ID, _TEXT_ID, _THINK_START_ID]) + + +def test_newline_before_think_end_dropped(parser): + text = "Line one.\nLine two.\n\nThe answer." + assert parser.extract_reasoning(text, None) == ( + "Line one.\nLine two.", + "\nThe answer.", + ) + + +def test_newline_before_think_end_dropped_streaming(parser): + reasoning, content = simulate_reasoning_streaming( + parser, + ["Line one.\n", "Line two.\n", "", "\nThe answer."], + [(_TEXT_ID,), (_TEXT_ID,), (_THINK_END_ID,), (_TEXT_ID,)], + ) + assert reasoning == "Line one.\nLine two." + assert content == "\nThe answer." + + +class TestPriorTurnReasoning: + """The template replays earlier assistant turns as + ``\\n...\\n\\n``, so a ```` from history must not end + reasoning for the new turn (#34211).""" + + _HISTORY = [ + _IM_START_ID, + _TEXT_ID, + _THINK_START_ID, + _TEXT_ID, + _THINK_END_ID, + _TEXT_ID, + _IM_END_ID, + ] + + def test_prior_turn_think_end_not_end(self, parser): + assert not parser.is_reasoning_end([*self._HISTORY, _IM_START_ID, _TEXT_ID]) + + def test_think_end_in_current_turn_is_end(self, parser): + assert parser.is_reasoning_end( + [*self._HISTORY, _IM_START_ID, _THINK_START_ID, _TEXT_ID, _THINK_END_ID] + ) + + +def test_reasoning_then_tool_call(parser, mock_request): + text = ( + "Need the weather.\n\n\n\n" + "\nTokyo\n\n\n" + ) + reasoning, content, tool_calls = parser.parse( + text, mock_request, enable_auto_tools=True + ) + assert reasoning == "Need the weather." + assert content is None + assert tool_calls is not None + assert tool_calls[0].name == "get_weather" + assert json.loads(tool_calls[0].arguments) == {"city": "Tokyo"} diff --git a/tests/parser/engine/trace_builder.py b/tests/parser/engine/trace_builder.py index 321ce6b9e0..3fa6e05078 100644 --- a/tests/parser/engine/trace_builder.py +++ b/tests/parser/engine/trace_builder.py @@ -46,6 +46,7 @@ from vllm.parser.engine.registered_adapters import ( Plamo3Parser, Qwen3Parser, SeedOssParser, + Step3p5Parser, ) # ── Data structures ────────────────────────────────────────────────── @@ -449,6 +450,19 @@ def _build_mimo(scenario: Scenario, validate: bool = True) -> Sample: ) +# ── Step-3.5 (Qwen3 XML, trailing reasoning whitespace stripped) ──── + + +def _build_step3p5(scenario: Scenario, validate: bool = True) -> Sample: + return _build_qwen3( + scenario, + name="step3p5", + parser_cls=Step3p5Parser, + strip_trailing_ws=True, + validate=validate, + ) + + # ── MiniMax M2 (XML invoke format, starts in REASONING) ────────────── _MINIMAX_M2_VOCAB: dict[str, int] = { @@ -1268,6 +1282,7 @@ _BUILDERS: dict[str, Any] = { "kimi_k2": _build_kimi_k2, "qwen3": _build_qwen3, "mimo": _build_mimo, + "step3p5": _build_step3p5, "inkling": _build_inkling, "plamo3": _build_plamo3, } diff --git a/tests/reasoning/test_step3p5_reasoning_parser.py b/tests/reasoning/test_step3p5_reasoning_parser.py index a65ac080f2..0d72250f1c 100644 --- a/tests/reasoning/test_step3p5_reasoning_parser.py +++ b/tests/reasoning/test_step3p5_reasoning_parser.py @@ -4,8 +4,8 @@ import pytest from tests.reasoning.utils import run_reasoning_extraction +from vllm.parser.engine.registered_adapters import Step3p5ParserReasoningAdapter from vllm.reasoning import ReasoningParserManager -from vllm.reasoning.step3p5_reasoning_parser import Step3p5ReasoningParser from vllm.tokenizers import get_tokenizer parser_name = "step3p5" @@ -112,7 +112,7 @@ EMPTY_STREAMING = { } NEW_LINE = { "output": "\nThis is a reasoning section\nThis is the rest", - "reasoning": "This is a reasoning section", + "reasoning": "\nThis is a reasoning section", "content": "\nThis is the rest", "is_reasoning_end": True, } @@ -126,7 +126,7 @@ NEW_LINE_STREAMING = { NEW_LINE_STREAMING_COMPLEX_CONTENT = { "output": "\n This is a \n reasoning section\n\n\n\n\nThis is the rest", - "reasoning": "\n This is a \n reasoning section\n\n", + "reasoning": "\n This is a \n reasoning section", "content": "\n\nThis is the rest", "is_reasoning_end": True, } @@ -285,7 +285,7 @@ def test_reasoning( step3p5_tokenizer.convert_tokens_to_string([token]) for token in output ] parser = ReasoningParserManager.get_reasoning_parser(parser_name)(step3p5_tokenizer) - assert isinstance(parser, Step3p5ReasoningParser) + assert isinstance(parser, Step3p5ParserReasoningAdapter) reasoning, content = run_reasoning_extraction( parser, output_tokens, streaming=streaming @@ -307,9 +307,10 @@ def test_reasoning( current_ids: list[int] = [] for token_id in output_ids: current_ids.append(token_id) - if token_id == parser.end_token_id: + if token_id in parser.reasoning_end_token_ids: assert parser.is_reasoning_end_streaming(current_ids, [token_id]) - assert parser.is_reasoning_end_streaming(output_ids, output_ids) + if param_dict["is_reasoning_end"]: + assert parser.is_reasoning_end_streaming(output_ids, output_ids) break assert not parser.is_reasoning_end_streaming(current_ids, [token_id]) else: @@ -333,6 +334,5 @@ def test_reasoning( step3p5_tokenizer.tokenize(param_dict["content"]) ) assert content_ids == expected_content_ids - else: - content_ids = parser.extract_content_ids(output_ids) - assert content_ids == [] + elif param_dict["is_reasoning_end"]: + assert parser.extract_content_ids(output_ids) == [] diff --git a/tests/tool_parsers/test_step3p5_tool_parser.py b/tests/tool_parsers/test_step3p5_tool_parser.py index 8291c62319..1e3111f5d3 100644 --- a/tests/tool_parsers/test_step3p5_tool_parser.py +++ b/tests/tool_parsers/test_step3p5_tool_parser.py @@ -847,7 +847,7 @@ TX assert chunk.tool_calls[0].function.name == "get_current_weather" assert chunk.tool_calls[0].type == "function" # Empty initially - assert chunk.tool_calls[0].function.arguments == "" + assert not chunk.tool_calls[0].function.arguments break assert header_found @@ -867,43 +867,6 @@ TX assert parsed_args["state"] == "TX" -def test_extract_tool_calls_complex_type_with_single_quote(step3p5_tokenizer): - """Test parameter type conversion based on tool schema.""" - tools = [ - ChatCompletionToolsParam( - type="function", - function={ - "name": "test_types", - "parameters": { - "type": "object", - "properties": { - "int_param": {"type": "integer"}, - "float_param": {"type": "float"}, - "bool_param": {"type": "boolean"}, - "str_param": {"type": "string"}, - "obj_param": {"type": "object"}, - }, - }, - }, - ) - ] - - model_output = """ - - -{'key': 'value'} - - -""" - - parser = Step3p5ToolParser(step3p5_tokenizer, tools=tools) - request = ChatCompletionRequest(model=MODEL, messages=[], tools=tools) - extracted_tool_calls = parser.extract_tool_calls(model_output, request=request) - - args = json.loads(extracted_tool_calls.tool_calls[0].function.arguments) - assert args["obj_param"] == {"key": "value"} - - def test_extract_tool_calls_streaming_mixed_content_and_multiple_tool_calls( step3p5_tool_parser, step3p5_tokenizer, sample_tools ): @@ -1016,8 +979,8 @@ def test_extract_tool_calls_non_streaming_mixed_content_and_multiple_tool_calls( """Test non-streaming extraction with mixed content and multiple tool calls. Scenario: Model outputs "hello" + complete tool call + "hi" + complete tool call. - Expected: "hello" as content, first tool call parsed (index=0), "hi" as content, - second tool call parsed (index=1) + Expected: "hello" as content, first tool call parsed (index=0), second tool call + parsed (index=1). Non-streaming content stops at the first tool call. """ # Model output: hello + complete tool call + hi + complete tool call model_output = """hello @@ -1066,17 +1029,7 @@ rectangle assert args_dict_1["dimensions"]["width"] == 10 assert args_dict_1["dimensions"]["height"] == 5 - # Verify content: should contain "hello", "hi" - assert extracted_tool_calls.content is not None - assert "hello" in extracted_tool_calls.content, "Should contain 'hello' as content" - assert "hi" in extracted_tool_calls.content, "Should contain 'hi' as content" - - # Verify the order: hello should come first, then hi - hello_index = extracted_tool_calls.content.find("hello") - hi_index = extracted_tool_calls.content.find("hi") - - assert hello_index >= 0, "'hello' should be in content" - assert hi_index > hello_index, "'hi' should come after 'hello'" + assert extracted_tool_calls.content == "hello" # Verify that tool call tags are NOT in the content assert "" not in extracted_tool_calls.content, ( @@ -1155,12 +1108,14 @@ rectangle request=request, ) - # Process delta result - if delta_result: - if delta_result.content: - other_content += delta_result.content - if delta_result.tool_calls: - for tool_call in delta_result.tool_calls: + # Serving flushes the parser at the end of the stream. + for delta in (delta_result, step3p5_tool_parser.finish_streaming()): + if not delta: + continue + if delta.content: + other_content += delta.content + if delta.tool_calls: + for tool_call in delta.tool_calls: idx = tool_call.index if idx not in tool_states: tool_states[idx] = { diff --git a/vllm/parser/engine/registered_adapters.py b/vllm/parser/engine/registered_adapters.py index 3c445c44c9..31295869c5 100644 --- a/vllm/parser/engine/registered_adapters.py +++ b/vllm/parser/engine/registered_adapters.py @@ -24,6 +24,7 @@ from vllm.parser.nemotron_v3 import NemotronV3Parser from vllm.parser.plamo3 import Plamo3Parser from vllm.parser.qwen3 import Qwen3Parser from vllm.parser.seed_oss import SeedOssParser +from vllm.parser.step3p5 import Step3p5Parser ( DeepSeekV32ParserReasoningAdapter, @@ -80,6 +81,11 @@ from vllm.parser.seed_oss import SeedOssParser SeedOssParserToolAdapter, ) = make_adapters(SeedOssParser) +( + Step3p5ParserReasoningAdapter, + Step3p5ParserToolAdapter, +) = make_adapters(Step3p5Parser) + ( Glm47MoeParserReasoningAdapter, Glm47MoeParserToolAdapter, diff --git a/vllm/parser/step3p5.py b/vllm/parser/step3p5.py new file mode 100644 index 0000000000..65b3212d1a --- /dev/null +++ b/vllm/parser/step3p5.py @@ -0,0 +1,49 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Step-3.5 / Step-3.7 parser for tool calls and reasoning. + +Both models use the Qwen3 XML tool-call grammar and ```` reasoning +unchanged. Two things differ from :class:`Qwen3Parser`: + +* The chat template always prefills ``\\n`` and has no switch to turn + thinking off, so ``enable_thinking`` is ignored. +* The model closes reasoning with ``\\n\\n``, so trailing reasoning + whitespace is stripped. +""" + +from __future__ import annotations + +import functools +from dataclasses import replace +from typing import TYPE_CHECKING + +from vllm.parser.engine.parser_engine_config import ParserEngineConfig +from vllm.parser.qwen3 import CHATML_TURN_BOUNDARIES, Qwen3Parser, qwen3_config + +if TYPE_CHECKING: + from vllm.tokenizers import TokenizerLike + from vllm.tool_parsers.abstract_tool_parser import Tool + + +@functools.cache +def step3p5_config() -> ParserEngineConfig: + return replace( + qwen3_config( + thinking=True, + name="step3p5", + turn_boundary_tokens=CHATML_TURN_BOUNDARIES, + ), + strip_trailing_reasoning_whitespace=True, + ) + + +class Step3p5Parser(Qwen3Parser): + def __init__( + self, + tokenizer: TokenizerLike, + tools: list[Tool] | None = None, + **kwargs, + ) -> None: + kwargs.setdefault("parser_engine_config", step3p5_config()) + super().__init__(tokenizer, tools, **kwargs) + self.thinking_enabled = True diff --git a/vllm/reasoning/__init__.py b/vllm/reasoning/__init__.py index 736a4fcdeb..9a4413cbd9 100644 --- a/vllm/reasoning/__init__.py +++ b/vllm/reasoning/__init__.py @@ -157,8 +157,8 @@ _REASONING_PARSERS_TO_REGISTER = { "Step3ReasoningParser", ), "step3p5": ( - "step3p5_reasoning_parser", - "Step3p5ReasoningParser", + "step3p5_engine_reasoning_parser", + "Step3p5ParserReasoningAdapter", ), "inkling": ( "inkling_reasoning_parser", diff --git a/vllm/reasoning/step3p5_engine_reasoning_parser.py b/vllm/reasoning/step3p5_engine_reasoning_parser.py new file mode 100644 index 0000000000..25e9546a1b --- /dev/null +++ b/vllm/reasoning/step3p5_engine_reasoning_parser.py @@ -0,0 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from vllm.parser.engine.registered_adapters import Step3p5ParserReasoningAdapter + +__all__ = ["Step3p5ParserReasoningAdapter"] diff --git a/vllm/reasoning/step3p5_reasoning_parser.py b/vllm/reasoning/step3p5_reasoning_parser.py deleted file mode 100644 index 5969ca5228..0000000000 --- a/vllm/reasoning/step3p5_reasoning_parser.py +++ /dev/null @@ -1,121 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project - -from collections.abc import Sequence -from typing import TYPE_CHECKING - -from vllm.entrypoints.generate.base.protocol import DeltaMessage -from vllm.reasoning.basic_parsers import BaseThinkingReasoningParser -from vllm.tokenizers import TokenizerLike - -if TYPE_CHECKING: - from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest - from vllm.entrypoints.openai.responses.protocol import ResponsesRequest - - -class Step3p5ReasoningParser(BaseThinkingReasoningParser): - """Reasoning parser for Step3p5 model. - - Step3p5 uses the ... format, but it tends to emit an extra - newline immediately before and/or after the token. This parser - trims the newline before in extract_reasoning*. - - Reasoning end detection doesn't wait for an extra token after so - structured output can constrain the first post-think token. - """ - - @property - def start_token(self) -> str: - return "" - - @property - def end_token(self) -> str: - return "" - - def __init__(self, tokenizer: TokenizerLike, *args, **kwargs): - super().__init__(tokenizer, *args, **kwargs) - - # Used to hold a trailing "\n" from reasoning content so we can decide - # whether it is immediately before . - self._pending_reasoning_newline = False - - def extract_reasoning( - self, - model_output: str, - request: "ChatCompletionRequest | ResponsesRequest", - ) -> tuple[str | None, str | None]: - reasoning, content = super().extract_reasoning(model_output, request) - if reasoning is not None: - reasoning = reasoning.removesuffix("\n") - return reasoning or None, content or None - - def extract_reasoning_streaming( - self, - previous_text: str, - current_text: str, - delta_text: str, - previous_token_ids: Sequence[int], - current_token_ids: Sequence[int], - delta_token_ids: Sequence[int], - ) -> DeltaMessage | None: - ret = super().extract_reasoning_streaming( - previous_text, - current_text, - delta_text, - previous_token_ids, - current_token_ids, - delta_token_ids, - ) - - if ret is None: - return None - - # Compatibility path for models that don't generate the start token: - # treat everything before as reasoning and everything after - # as content. - if ( - self.start_token_id not in previous_token_ids - and self.start_token_id not in delta_token_ids - ): - if self.end_token_id in delta_token_ids: - end_index = delta_text.find(self.end_token) - reasoning = delta_text[:end_index] - content = delta_text[end_index + len(self.end_token) :] - ret = DeltaMessage(reasoning=reasoning, content=content or None) - elif self.end_token_id in previous_token_ids: - ret = DeltaMessage(content=delta_text) - else: - ret = DeltaMessage(reasoning=delta_text) - - reasoning_to_output = ret.reasoning - content_to_output = ret.content - - # Reasoning: handle the newline immediately before . - if reasoning_to_output is not None: - if self._pending_reasoning_newline: - reasoning_to_output = "\n" + reasoning_to_output - self._pending_reasoning_newline = False - - if reasoning_to_output.endswith("\n"): - reasoning_to_output = reasoning_to_output.removesuffix("\n") - if self.end_token in delta_text: - # Trailing "\n" is right before , drop it. - self._pending_reasoning_newline = False - else: - # Hold the trailing "\n" until we know whether follows. - self._pending_reasoning_newline = True - - # Content: handle the newline immediately after . - if content_to_output is not None: - # If we have content, reasoning must have ended. - self._pending_reasoning_newline = False - - if self.end_token in delta_text and content_to_output.startswith("\n"): - content_to_output = content_to_output.removeprefix("\n") - - reasoning_to_output = reasoning_to_output or None - content_to_output = content_to_output or None - if reasoning_to_output is None and content_to_output is None: - return None - - return DeltaMessage(reasoning=reasoning_to_output, content=content_to_output) diff --git a/vllm/tool_parsers/step3p5_tool_parser.py b/vllm/tool_parsers/step3p5_tool_parser.py index ca0d86a9e0..9d964ce314 100644 --- a/vllm/tool_parsers/step3p5_tool_parser.py +++ b/vllm/tool_parsers/step3p5_tool_parser.py @@ -1,1464 +1,8 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project -import json -from collections.abc import Sequence -from typing import Any -from xml.parsers.expat import ParserCreate -import regex as re +from vllm.parser.engine.registered_adapters import Step3p5ParserToolAdapter -from vllm.entrypoints.chat_utils import make_tool_call_id -from vllm.entrypoints.generate.base.protocol import ( - DeltaFunctionCall, - DeltaMessage, - DeltaToolCall, - ExtractedToolCallInformation, - FunctionCall, - ToolCall, -) -from vllm.entrypoints.openai.chat_completion.protocol import ( - ChatCompletionRequest, -) -from vllm.logger import init_logger -from vllm.tokenizers import TokenizerLike -from vllm.tool_parsers.abstract_tool_parser import Tool, ToolParser -from vllm.tool_parsers.utils import ( - find_tool_name, - find_tool_properties, - safe_literal_eval, -) -logger = init_logger(__name__) - - -class StreamingXMLToolCallParser: - """Simplified streaming XML tool call parser - Supports streaming input, parsing, and output - """ - - def __init__(self): - self.reset_streaming_state() - - # Tool configuration information - self.tools: list[Tool] | None = None - self.tool_call_start_token: str = "" - self.tool_call_end_token: str = "" - self.function_start_token: str = " DeltaMessage: - """Parse single streaming XML chunk and return Delta response - This is the actual streaming interface that receives chunks - one by one and maintains internal state - - Args: - xml_chunk: Single XML chunk string - Returns: - DeltaMessage: Contains delta information generated by this chunk, - returns empty response if no complete elements - - """ - # Record delta count before processing - initial_delta_count = len(self.deltas) - entry_call_id = self.current_call_id - entry_tool_call_index = self.tool_call_index - - self.streaming_buffer += xml_chunk - - found_elements = self._process_complete_xml_elements() - - fallback_call_id = None - if entry_call_id is not None: - if ( - self.current_call_id == entry_call_id - and self.tool_call_index == entry_tool_call_index - ): - fallback_call_id = entry_call_id - elif ( - self.current_call_id is not None - and self.tool_call_index == entry_tool_call_index + 1 - ): - fallback_call_id = self.current_call_id - - if found_elements: - # If complete elements found, check if end events were missed - # some tags may not have been triggered - try: - new_deltas = self.deltas[initial_delta_count:] - # If this chunk contains - # but didn't generate '}', then complete it - if ( - fallback_call_id is not None - and self.function_end_token in xml_chunk - ): - # - Added '}' (non-empty parameter ending) - # - Added '{}' (empty parameter function) - has_function_close = any( - ( - td.tool_calls - and any( - ( - tc.function - and tc.id == fallback_call_id - and isinstance(tc.function.arguments, str) - and (tc.function.arguments in ("}", "{}")) - ) - for tc in td.tool_calls - ) - ) - for td in new_deltas - ) - if not has_function_close: - # Close potentially unclosed element - if self.current_param_name: - self._end_element("parameter") - if self.current_function_name: - self._end_element("function") - # If this chunk contains - # but didn't generate final empty delta, then complete it - if ( - fallback_call_id is not None - and self.tool_call_end_token in xml_chunk - ): - has_toolcall_close = any( - ( - td.tool_calls - and any( - ( - tc.type == "function" - and tc.function - and tc.function.arguments == "" - and tc.id == fallback_call_id - ) - for tc in td.tool_calls - ) - ) - for td in new_deltas - ) - if not has_toolcall_close: - # Close potentially unclosed element - if self.current_param_name: - self._end_element("parameter") - if self.current_function_name: - self._end_element("function") - self._end_element("tool_call") - except Exception as e: - logger.warning("Error with fallback parsing: %s", e) - # Merge newly generated deltas into single response - result_delta = self._merge_new_deltas_to_single_response( - initial_delta_count - ) - return result_delta - else: - # No complete elements, check if there's unoutput text content - if self.text_content_buffer and self.tool_call_index == 0: - # Has text content but no tool_call yet, output text content - text_delta = DeltaMessage(content=self.text_content_buffer) - self._emit_delta(text_delta) - # Clear buffer to avoid duplicate output - self.text_content_buffer = "" - return text_delta - - # If this chunk contains end tags but wasn't triggered by parser, - # manually complete end events - # Only execute when still on the same call as when entered, - # to prevent accidentally closing new calls - # in multi scenarios - if fallback_call_id is not None and ( - self.function_end_token in xml_chunk - or self.tool_call_end_token in xml_chunk - ): - # Close potentially unclosed element - if self.current_param_name: - self._end_element("parameter") - if self.function_end_token in xml_chunk and self.current_function_name: - self._end_element("function") - if self.tool_call_end_token in xml_chunk: - self._end_element("tool_call") - # Return the merged delta result generated by this fallback - result_delta = self._merge_new_deltas_to_single_response( - initial_delta_count - ) - return result_delta - - # No complete elements, return empty response - return DeltaMessage(content=None) - - def _escape_xml_special_chars(self, text: str) -> str: - """Escape XML special characters - Args: - text: Original text - Returns: - Escaped text - """ - xml_escapes = { - "&": "&", - "<": "<", - ">": ">", - '"': """, - "'": "'", - } - - for char, escape in xml_escapes.items(): - text = text.replace(char, escape) - - return text - - def _process_complete_xml_elements(self) -> bool: - """Process complete XML elements in buffer. - - Returns: - bool: Whether complete elements were found and processed - - """ - found_any = False - - while self.last_processed_pos < len(self.streaming_buffer): - # Find next complete xml element - element, end_pos = self._find_next_complete_element(self.last_processed_pos) - if element is None: - # No complete element found, wait for more data - break - - # Check if this element should be skipped - if self._should_skip_element(element): - self.last_processed_pos = end_pos - continue - - # Found complete XML element, process it - try: - preprocessed_element = self._preprocess_xml_chunk(element) - # Check if this is the first tool_call start - if ( - ( - preprocessed_element.strip().startswith("") - or preprocessed_element.strip().startswith("") - and self.tool_call_index > 0 - and self.current_call_id - and self.current_function_name - ): - # Reset parser state but preserve generated deltas - if self.current_param_name: - self._end_element("parameter") - if self.current_function_open: - self._end_element("function") - # Output final tool_call tail delta - final_delta = DeltaMessage( - role=None, - content=None, - reasoning=None, - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments=""), - ) - ], - ) - self._emit_delta(final_delta) - # Reset XML parser and current call state - self._reset_xml_parser_after_tool_call() - # Parse preprocessed element - self.parser.Parse(preprocessed_element, False) - found_any = True - - except Exception as e: - logger.warning("Error when parsing XML elements: %s", e) - - # Update processed position - self.last_processed_pos = end_pos - - return found_any - - def _fix_incomplete_tag_in_chunk(self, chunk: str) -> str: - """Fallback: fix incomplete ) - Examples: , - Also handles missing = cases: -> , - -> - Only fixes tags that pass validation (parameter exists in tool definition) - """ - # First, handle missing = cases for function tags - chunk = self._fix_missing_equals_in_function_tag(chunk) - - for tag_type in ["parameter", "function"]: - pattern = f"<{tag_type}=" - if pattern not in chunk: - continue - - start_idx = chunk.find(pattern) - after_tag = chunk[start_idx:] - gt_pos = after_tag.find(">") - lt_pos = after_tag.find("<", len(pattern)) - - # Skip if already well-formed - if ( - gt_pos != -1 - and (lt_pos == -1 or gt_pos < lt_pos) - and pattern in after_tag[:gt_pos] - ): - continue - - # Extract tag name (stop at space, newline, or <) - content = chunk[start_idx + len(pattern) :] - end_pos = next( - (i for i, ch in enumerate(content) if ch in (" ", "\n", "<")), - len(content), - ) - tag_name = content[:end_pos] - - if not tag_name: - continue - - # Remove duplicate prefix: ", 1 - ) - - return chunk - - def _fix_missing_equals_in_function_tag(self, chunk: str) -> str: - """Fix missing = in function tags: or - Examples: - -> - -> - Only fixes if function name exists in tool definition - """ - # already correct - if " (with space/newline but no =) - pattern1 = r"" - match1 = re.search(pattern1, chunk) - if match1: - func_name = match1.group(1).strip() - # must validate function name exists before fixing - if func_name and self._validate_function_name(func_name): - original = match1.group(0) - fixed = f"" - chunk = chunk.replace(original, fixed, 1) - return chunk - - # Pattern 2: (no space, no =) - # only match " - match2 = re.search(pattern2, chunk) - if match2: - func_name = match2.group(1).strip() - # must validate function name exists before fixing - if func_name and self._validate_function_name(func_name): - original = match2.group(0) - fixed = f"" - chunk = chunk.replace(original, fixed, 1) - return chunk - - return chunk - - def _validate_function_name(self, func_name: str) -> bool: - """Check if function name exists in tool definitions.""" - return find_tool_name(self.tools, func_name) - - def _validate_parameter_name(self, param_name: str) -> bool: - """Check if parameter exists in current function's tool definition.""" - if not self.tools or not self.current_function_name: - return True - properties = find_tool_properties(self.tools, self.current_function_name) - return not properties or param_name in properties - - def _should_skip_element(self, element: str) -> bool: - """Determine whether an element should be skipped. - - Args: - element: Element to evaluate - - Returns: - bool: True means should skip, False means should process - - """ - # If it's a tool_call XML tag, don't skip - if ( - element.startswith(self.tool_call_start_token) - or element.startswith(self.function_start_token) - or element.startswith(self.parameter_start_token) - ): - return False - - # If currently not parsing tool calls and not blank, - # collect this text instead of skipping - # Only process other XML elements after tool_call appears, - # otherwise treat as plain text - if self.current_call_id is None and element: - # Collect text content to buffer - self.text_content_buffer += element - return True # Still skip, but content has been collected - - # If currently parsing tool calls, - # this might be parameter value, don't skip - if self.current_call_id is not None: - return False - - # Skip blank content - return not element - - def _find_next_complete_element(self, start_pos: int) -> tuple[str | None, int]: - """Find next complete XML element from specified position. - - Args: - start_pos: Position to start searching - - Returns: - (Complete element string, element end position), - returns (None, start_pos) if no complete element found - - """ - buffer = self.streaming_buffer[start_pos:] - - if not buffer: - return None, start_pos - - if buffer.startswith("<"): - # Check if this is an incomplete parameter/function tag - # e.g., " not in buffer.split("\n")[0] - ) - is_incomplete_func = ( - buffer.startswith("" not in buffer.split("\n")[0] - ) - - if is_incomplete_param or is_incomplete_func: - # Find the corresponding closing tag - tag_type = "parameter" if is_incomplete_param else "function" - closing_tag = f"" - closing_pos = buffer.find(closing_tag) - - if closing_pos != -1: - # Found closing tag, return complete element including closing tag - complete_element = buffer[: closing_pos + len(closing_tag)] - return complete_element, start_pos + closing_pos + len(closing_tag) - - # Need to ensure no new < appears, - # find the nearest one between < and > - tag_end = buffer.find("<", 1) - tag_end2 = buffer.find(">", 1) - if tag_end != -1 and tag_end2 != -1: - # Next nearest is < - if tag_end < tag_end2: - return buffer[:tag_end], start_pos + tag_end - # Next nearest is >, means found XML element - else: - return buffer[: tag_end2 + 1], start_pos + tag_end2 + 1 - elif tag_end != -1: - return buffer[:tag_end], start_pos + tag_end - elif tag_end2 != -1: - return buffer[: tag_end2 + 1], start_pos + tag_end2 + 1 - else: - # If currently not parsing tool calls (entering a tool_call), - # check if starts with or - if buffer == ""[: len(buffer)]: - # Might be start of , wait for more data - return None, start_pos - elif ( - buffer.startswith(" DeltaMessage: - """Merge newly generated deltas from this processing - into a single DeltaMessage - - Args: - initial_count: Delta count before processing - - Returns: - Merged DeltaMessage containing all newly generated delta information - - """ - if len(self.deltas) <= initial_count: - return DeltaMessage(content=None) - - # Get newly generated deltas - new_deltas = self.deltas[initial_count:] - - if len(new_deltas) == 1: - # Only one new delta, return directly - return new_deltas[0] - - # Merge multiple new deltas - merged_tool_calls: list[DeltaToolCall] = [] - merged_content: str = "" - - for delta in new_deltas: - if delta.content: - merged_content += delta.content - if delta.tool_calls: - # For tool_calls, we need to intelligently merge arguments - for tool_call in delta.tool_calls: - # Find if there's already a tool_call with the same call_id - existing_call = None - for existing in merged_tool_calls: - if existing.id == tool_call.id: - existing_call = existing - break - - if existing_call and existing_call.function: - # Merge to existing tool_call - if tool_call.function and tool_call.function.name: - existing_call.function.name = tool_call.function.name - if ( - tool_call.function - and tool_call.function.arguments is not None - ): - if existing_call.function.arguments is None: - existing_call.function.arguments = "" - - # For streaming JSON parameters, - # simply concatenate in order - new_args = tool_call.function.arguments - existing_call.function.arguments += new_args - if tool_call.type: - existing_call.type = tool_call.type - else: - # Add new tool_call - merged_tool_calls.append(tool_call) - - return DeltaMessage( - content=merged_content if merged_content else None, - tool_calls=merged_tool_calls, - ) - - def _preprocess_xml_chunk(self, chunk: str) -> str: - """Preprocess XML chunk, handle non-standard formats, - and escape special characters - - Args: - chunk: Original XML chunk - - Returns: - Processed XML chunk - - """ - # Check if this is a tool_call related element - is_tool_call = False - if chunk.startswith(self.tool_call_start_token) or chunk.startswith( - self.tool_call_end_token - ): - is_tool_call = True - # Check for function tags (including malformed ones without =) - # , , , - if ( - chunk.startswith(self.function_start_token) - or chunk.startswith(self.function_end_token) - or chunk.startswith(" - # This handles cases like: format -> - processed = re.sub(r"]+)>", r'', chunk) - # Handle format -> - processed = re.sub(r"]+)>", r'', processed) - - original_chunk = chunk - # If in parameter value accumulation mode - if self._pre_inside_parameter: - # Parameter end: output accumulated raw text - # safely then return - if processed.startswith(""): - body_text = self._pre_param_buffer - # Trigger deferred parsing mode - # literal_eval+json output in end_element - self.defer_current_parameter = True - self.deferred_param_raw_value = body_text - # Clean up state - self._pre_inside_parameter = False - self._pre_param_buffer = "" - self._pre_current_param_name = None - safe_text = self._escape_xml_special_chars(body_text) - return f"{safe_text}" - else: - # If this is the first block of content after entering parameter - # evaluate if deferred parsing is needed; - # If not needed, exit accumulation mode - # and pass through directly - if self._pre_param_buffer == "": - # Get current parameter type - param_type = ( - self._get_param_type(self._pre_current_param_name) - if self._pre_current_param_name - else "string" - ) - # Only these types need deferred parsing to - # handle Python literals containing single quotes - is_object_type = param_type in ["object"] - is_complex_type = ( - param_type in ["array", "arr", "sequence"] - or param_type.startswith("dict") - or param_type.startswith("list") - ) - - # Only delay when contains container symbols - # and has single quotes and is complex type - has_container_hint = ( - ("[" in original_chunk) - or ("{" in original_chunk) - or ("(" in original_chunk) - ) - - # Determine if deferred parsing is needed - need_defer = False - if is_complex_type: - # Complex type, always need deferred parsing - need_defer = True - elif ( - is_object_type - and has_container_hint - and ("'" in original_chunk) - ): - # Object type with container symbols - # and single quotes, need deferred parsing - need_defer = True - - if not need_defer: - # No need for deferred parsing, - # exit parameter mode directly - self._pre_inside_parameter = False - return self._escape_xml_special_chars(original_chunk) - self._pre_param_buffer += original_chunk - return "" - - # Parameter start: enable accumulation - if processed.startswith("', processed) - if m: - self._pre_current_param_name = m.group(1) - self._pre_inside_parameter = True - self._pre_param_buffer = "" - return processed - - # If processed doesn't contain special_token, escape processed - # This is because XML parsing encounters special characters - # and reports errors, so escaping is needed - if not is_tool_call: - processed = self._escape_xml_special_chars(processed) - return processed - - def _emit_delta(self, delta: DeltaMessage): - """Emit Delta response (streaming output).""" - self.deltas.append(delta) - - def _auto_close_open_parameter_if_needed(self, incoming_tag: str | None = None): - """Before starting to process new elements, - if there are unclosed tags from before, - automatically complete their endings to the parser. - - If there are unclosed parameters, - it's equivalent to feeding `` - - When about to start a new function or tool_call, - if there are unclosed functions, complete ``. - - When about to start a new tool_call, - if there are unclosed tool_calls, complete ``. - """ - # First close unclosed parameters - if self.current_param_name: - self._end_element("parameter") - - # If about to start new function or tool_call, - # and there are unclosed functions, close function first - if incoming_tag in ("function", "tool_call") and self.current_function_name: - self._end_element("function") - - # If about to start new tool_call, - # and there are unclosed tool_calls, close tool_call first - if incoming_tag == "tool_call" and self.current_call_id: - self._end_element("tool_call") - - def _start_element(self, name: str, attrs: dict[str, str]): - """Handle XML start element events.""" - if name == "root": - return - - if name == "tool_call": - # Before opening new tool_call, - # automatically complete previous unclosed tags - self._auto_close_open_parameter_if_needed("tool_call") - - self.parameters = {} - self.current_call_id = make_tool_call_id() - self.current_param_is_first = True - self.tool_call_index += 1 - elif name.startswith("function") or (name == "function"): - # If missing tool_call, manually complete - if not self.current_call_id: - self._start_element("tool_call", {}) - # Before opening new function, - # automatically complete previous unclosed tags (parameter/function) - self._auto_close_open_parameter_if_needed("function") - function_name = self._extract_function_name(name, attrs) - self.current_function_name = function_name - self.current_function_open = True - if function_name: - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall( - name=function_name, arguments="" - ), - ) - ] - ) - self._emit_delta(delta) - elif name.startswith("parameter") or (name == "parameter"): - # If previous parameter hasn't ended normally, - # complete its end first, then start new parameter - self._auto_close_open_parameter_if_needed("parameter") - param_name = self._extract_parameter_name(name, attrs) - self.current_param_name = param_name - self.current_param_value = "" - self.current_param_value_converted = "" - self.start_quote_emitted = False # Reset start quote flag - - # Only output parameter name and colon, - # don't output quotes - # decide after parameter value type is determined - if param_name: - if not self.parameters: - # First parameter - # start JSON, only output parameter name and colon - json_start = f'{{"{param_name}": ' - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall( - name=None, arguments=json_start - ), - ) - ] - ) - self._emit_delta(delta) - self.current_param_is_first = True - else: - # Subsequent parameters - # add comma and parameter name, no quotes - json_continue = f', "{param_name}": ' - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall( - name=None, arguments=json_continue - ), - ) - ] - ) - self._emit_delta(delta) - self.current_param_is_first = False - - def _char_data(self, data: str): - """Handle XML character data events.""" - if data and self.current_param_name: - # If preprocessing stage determines deferred parsing is needed, - # only cache character data, no streaming output - if self.defer_current_parameter: - original_data = data - if self.should_emit_end_newline: - original_data = "\n" + original_data - self.should_emit_end_newline = False - if original_data.endswith("\n"): - self.should_emit_end_newline = True - original_data = original_data[:-1] - self.current_param_value += original_data - return - - param_type = self._get_param_type(self.current_param_name) - - # Check if this is the first time receiving data for this parameter - # If this is the first packet of data and starts with \n, remove \n - if not self.current_param_value and data.startswith("\n"): - data = data[1:] - - # Output start quote for string type (if not already output) - if ( - param_type in ["string", "str", "text", "varchar", "char", "enum"] - and not self.start_quote_emitted - ): - quote_delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments='"'), - ) - ] - ) - self._emit_delta(quote_delta) - self.start_quote_emitted = True - - if not data: - return - - original_data = data - # Delay output of trailing newline - if self.should_emit_end_newline: - original_data = "\n" + original_data - self.should_emit_end_newline = False - if original_data.endswith("\n"): - self.should_emit_end_newline = True - original_data = original_data[:-1] - self.current_param_value += original_data - - # convert parameter value by param_type - converted_value = self._convert_param_value( - self.current_param_value, param_type - ) - output_data = self._convert_for_json_streaming(converted_value, param_type) - - delta_data = output_data[len(self.current_param_value_converted) :] - self.current_param_value_converted = output_data - - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments=delta_data), - ) - ] - ) - self._emit_delta(delta) - - def _end_element(self, name: str): - """Handle XML end element events.""" - if name == "root": - return - - # If function or tool_call ends and there are still unclosed parameters, - # complete parameter end first - if ( - name.startswith("function") or name == "function" or name == "tool_call" - ) and self.current_param_name: - self._auto_close_open_parameter_if_needed() - - if ( - name.startswith("parameter") or name == "parameter" - ) and self.current_param_name: - # End current parameter - param_name = self.current_param_name - param_value = self.current_param_value - - # If in deferred parsing mode, - # perform overall parsing on raw content - # accumulated in preprocessing stage and output once - if self.defer_current_parameter: - raw_text = ( - self.deferred_param_raw_value - if self.deferred_param_raw_value - else param_value - ) - parsed_value = None - output_arguments = None - try: - # If previously delayed trailing newline, - # add it back before parsing - if self.should_emit_end_newline: - raw_for_parse = raw_text + "\n" - else: - raw_for_parse = raw_text - parsed_value = safe_literal_eval(raw_for_parse) - output_arguments = json.dumps(parsed_value, ensure_ascii=False) - except Exception: - # Fallback: output as string as-is - output_arguments = json.dumps(raw_text, ensure_ascii=False) - parsed_value = raw_text - - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall( - name=None, arguments=output_arguments - ), - ) - ] - ) - self._emit_delta(delta) - - # Clean up and store - self.should_emit_end_newline = False - self.parameters[param_name] = parsed_value - self.current_param_name = None - self.current_param_value = "" - self.current_param_value_converted = "" - self.start_quote_emitted = False - self.defer_current_parameter = False - self.deferred_param_raw_value = "" - return - - param_type = self._get_param_type(param_name) - - # convert complete parameter value by param_type - converted_value = self._convert_param_value(param_value, param_type) - - # Decide whether to add end quote based on parameter type - if param_type in ["string", "str", "text", "varchar", "char", "enum"]: - # For empty string parameters, need special handling - if not param_value and not self.start_quote_emitted: - # No start quote output, - # directly output complete empty string - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments='""'), - ) - ] - ) - self._emit_delta(delta) - else: - # Non-empty parameter value, output end quote - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments='"'), - ) - ] - ) - self._emit_delta(delta) - - self.should_emit_end_newline = False - # Store converted value - self.parameters[param_name] = converted_value - self.current_param_name = None - self.current_param_value = "" - self.current_param_value_converted = "" - self.start_quote_emitted = False - - elif name.startswith("function") or name == "function": - # if there are parameters, close JSON object - if self.parameters: - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments="}"), - ) - ] - ) - self._emit_delta(delta) - # return empty object - else: - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments="{}"), - ) - ] - ) - self._emit_delta(delta) - self.current_function_open = False - self.current_function_name = ( - None # Clear function name to prevent duplicate closing - ) - - elif name == "tool_call": - # Before ending tool_call, - # ensure function is closed to complete missing right brace - if self.current_function_open: - # If there are still unclosed parameters, close them first - if self.current_param_name: - self._end_element("parameter") - # Close function, ensure output '}' or '{}' - self._end_element("function") - # Final Delta - delta = DeltaMessage( - tool_calls=[ - DeltaToolCall( - index=self.tool_call_index - 1, - id=self.current_call_id, - type="function", - function=DeltaFunctionCall(name=None, arguments=""), - ) - ] - ) - self._emit_delta(delta) - - # Check if there's text content to output (between tool_calls) - if self.text_content_buffer.strip(): - text_delta = DeltaMessage(content=self.text_content_buffer) - self._emit_delta(text_delta) - - self._reset_xml_parser_after_tool_call() - - def setup_parser(self): - """Set up XML parser event handlers.""" - self.parser.buffer_text = True - self.parser.StartElementHandler = self._start_element - self.parser.EndElementHandler = self._end_element - self.parser.CharacterDataHandler = self._char_data - - def set_tools(self, tools: list[Tool] | None): - """Set tool configuration information.""" - self.tools = tools - - def _extract_function_name(self, name: str, attrs: dict[str, str]) -> str | None: - """Extract function name from various formats.""" - if attrs and "name" in attrs: - return attrs["name"] - - if "=" in name: - parts = name.split("=", 1) - if len(parts) == 2 and parts[0] == "function": - return parts[1] - - return None - - def _extract_parameter_name(self, name: str, attrs: dict[str, str]) -> str | None: - """Extract parameter name from various formats.""" - if attrs and "name" in attrs: - return attrs["name"] - - if "=" in name: - parts = name.split("=", 1) - if len(parts) == 2 and parts[0] == "parameter": - return parts[1] - - return None - - def _get_param_type(self, param_name: str) -> str: - """Get parameter type based on tool configuration, defaults to string - Args: - param_name: Parameter name - - Returns: - Parameter type - - """ - if not self.tools or not self.current_function_name: - return "string" - properties = find_tool_properties(self.tools, self.current_function_name) - param_config = properties.get(param_name) - if isinstance(param_config, dict): - return self.repair_param_type(str(param_config.get("type", "string"))) - return "string" - - def repair_param_type(self, param_type: str) -> str: - """Repair unknown parameter types by treating them as string - Args: - param_type: Parameter type - - Returns: - Repaired parameter type - - """ - if ( - param_type in ["string", "str", "text", "varchar", "char", "enum"] - or param_type.startswith("int") - or param_type.startswith("uint") - or param_type.startswith("long") - or param_type.startswith("short") - or param_type.startswith("unsigned") - or param_type.startswith("num") - or param_type.startswith("float") - or param_type in ["boolean", "bool", "binary"] - or ( - param_type in ["object", "array", "arr", "sequence"] - or param_type.startswith("dict") - or param_type.startswith("list") - ) - ): - return param_type - else: - return "string" - - def _convert_param_value(self, param_value: str, param_type: str) -> Any: - """Convert value based on parameter type - Args: - param_value: Parameter value - param_type: Parameter type - - Returns: - Converted value - - """ - if param_value.lower() == "null": - return None - - param_type = param_type.strip().lower() - if param_type in ["string", "str", "text", "varchar", "char", "enum"]: - return param_value - elif ( - param_type.startswith("int") - or param_type.startswith("uint") - or param_type.startswith("long") - or param_type.startswith("short") - or param_type.startswith("unsigned") - ): - try: - return int(param_value) - except (ValueError, TypeError): - logger.warning( - "Parsed value '%s' is not an integer, degenerating to string.", - param_value, - ) - return param_value - elif param_type.startswith("num") or param_type.startswith("float"): - try: - float_param_value: float = float(param_value) - return ( - float_param_value - if float_param_value - int(float_param_value) != 0 - else int(float_param_value) - ) - except (ValueError, TypeError): - logger.warning( - "Parsed value '%s' is not a float, degenerating to string.", - param_value, - ) - return param_value - elif param_type in ["boolean", "bool", "binary"]: - param_value = param_value.lower() - return param_value == "true" - else: - return param_value - - def _convert_for_json_streaming(self, converted_value: Any, param_type: str) -> str: - """Convert converted_value based on - whether it's empty and if type is string - Args: - converted_value: Converted value - param_type: Parameter type - - Returns: - Converted string for streaming output - - """ - # Check if value is empty, but exclude numeric 0 - if converted_value is None or converted_value == "": - return "" - - if param_type in ["string", "str", "text", "varchar", "char", "enum"]: - # String type, remove double quotes - return json.dumps(converted_value, ensure_ascii=False)[1:-1] - else: - # Non-string type, return complete JSON string - if not isinstance(converted_value, str): - return json.dumps(converted_value, ensure_ascii=False) - else: - return converted_value - - def _reset_xml_parser_after_tool_call(self): - """Each tool_call is treated as a separate XML document, - so we need to reset the parser after each tool_call. - """ - # recreate XML parser - self.parser = ParserCreate() - self.setup_parser() - - # Reset current tool_call state - if self.current_call_id: - self.last_completed_call_id = self.current_call_id - self.current_call_id = None - self.current_function_name = None - self.current_function_open = False - self.parameters = {} - self.current_param_name = None - self.current_param_value = "" - self.current_param_value_converted = "" - self.current_param_is_first = False - self.should_emit_end_newline = False - self.start_quote_emitted = False - self.text_content_buffer = "" - - # Reset preprocessing and deferred parsing state - self._pre_inside_parameter = False - self._pre_param_buffer = "" - self._pre_current_param_name = None - self.defer_current_parameter = False - self.deferred_param_raw_value = "" - - -class Step3p5ToolParser(ToolParser): - # Step-3.5/3.7 chat templates use the Qwen3-Coder XML tool-call format. +class Step3p5ToolParser(Step3p5ParserToolAdapter): # type: ignore[valid-type, misc] structural_tag_model = "qwen_3_coder" - - def __init__(self, tokenizer: TokenizerLike, tools: list[Tool] | None = None): - super().__init__(tokenizer, tools) - self.parser = StreamingXMLToolCallParser() - - # Add missing attributes for compatibility with serving_chat.py - self.prev_tool_call_arr: list[dict] = [] - self.streamed_args_for_tool: list[str] = [] - - logger.info( - "vLLM Successfully import tool parser %s !", self.__class__.__name__ - ) - - def extract_tool_calls( - self, - model_output: str, - request: ChatCompletionRequest, - ) -> ExtractedToolCallInformation: - self.parser.reset_streaming_state() - # Reset tool call tracking arrays for new extraction - self.prev_tool_call_arr = [] - self.streamed_args_for_tool = [] - self.parser.set_tools(self.tools) - result = self.parser.parse_single_streaming_chunks(model_output) - if not result.tool_calls: - return ExtractedToolCallInformation( - tool_calls=[], - tools_called=False, - content=result.content, - ) - else: - tool_calls = [] - for tool_call in result.tool_calls: - if tool_call.function and tool_call.function.name: - tool_calls.append( - ToolCall( - id=tool_call.id, - type=tool_call.type, - function=FunctionCall( - name=tool_call.function.name, - arguments=tool_call.function.arguments, - ), - ) - ) - - # Update tool call tracking arrays for compatibility - tool_index = ( - tool_call.index - if tool_call.index is not None - else len(self.prev_tool_call_arr) - 1 - ) - - # Ensure we have enough entries in our tracking arrays - while len(self.prev_tool_call_arr) <= tool_index: - self.prev_tool_call_arr.append({"name": "", "arguments": ""}) - while len(self.streamed_args_for_tool) <= tool_index: - self.streamed_args_for_tool.append("") - - # Update tool call information - self.prev_tool_call_arr[tool_index]["name"] = ( - tool_call.function.name - ) - self.prev_tool_call_arr[tool_index]["arguments"] = ( - tool_call.function.arguments - ) - - # Update streamed arguments - if tool_call.function.arguments: - self.streamed_args_for_tool[tool_index] = ( - tool_call.function.arguments - ) - - return ExtractedToolCallInformation( - tool_calls=tool_calls, - tools_called=len(tool_calls) > 0, - content=result.content, - ) - - def extract_tool_calls_streaming( - self, - previous_text: str, - current_text: str, - delta_text: str, - previous_token_ids: Sequence[int], - current_token_ids: Sequence[int], - delta_token_ids: Sequence[int], - request: ChatCompletionRequest, - ) -> DeltaMessage | None: - if not previous_text: - self.parser.reset_streaming_state() - # Reset tool call tracking arrays for new streaming session - self.prev_tool_call_arr = [] - self.streamed_args_for_tool = [] - self.parser.set_tools(self.tools) - - # Model sometimes outputs separately causing delta_text to be empty. - # If there were tool_calls before and all current tool_calls have ended, - # return an empty tool_call for outer streaming output - # to correctly output tool_call field - if not delta_text and delta_token_ids: - open_calls = current_text.count( - self.parser.tool_call_start_token - ) - current_text.count(self.parser.tool_call_end_token) - if ( - open_calls == 0 - and self.parser.tool_call_index > 0 - or not self.parser.tool_call_index - and current_text - ): - return DeltaMessage(content="") - return None - - # Parse the delta text and get the result - result = self.parser.parse_single_streaming_chunks(delta_text) - - # Update tool call tracking arrays based on incremental parsing results - if result and result.tool_calls: - for tool_call in result.tool_calls: - if tool_call.function: - tool_index = ( - tool_call.index - if tool_call.index is not None - else len(self.prev_tool_call_arr) - 1 - ) - - # Ensure we have enough entries in our tracking arrays - while len(self.prev_tool_call_arr) <= tool_index: - self.prev_tool_call_arr.append({"name": "", "arguments": ""}) - while len(self.streamed_args_for_tool) <= tool_index: - self.streamed_args_for_tool.append("") - - # Update tool name if provided - if tool_call.function.name: - self.prev_tool_call_arr[tool_index]["name"] = ( - tool_call.function.name - ) - - # Update arguments incrementally - if tool_call.function.arguments is not None: - # Concatenate the incremental arguments - # to the existing streamed arguments - self.prev_tool_call_arr[tool_index]["arguments"] += ( - tool_call.function.arguments - ) - self.streamed_args_for_tool[tool_index] += ( - tool_call.function.arguments - ) - return result - - def parser_should_check_for_unstreamed_tool_arg_tokens(self) -> bool: - """Skip the remaining_call calculation in serving_chat.""" - return False