diff --git a/python/sglang/srt/entrypoints/openai/serving_chat.py b/python/sglang/srt/entrypoints/openai/serving_chat.py index dcf549aa7dfe..a1a394cbec23 100644 --- a/python/sglang/srt/entrypoints/openai/serving_chat.py +++ b/python/sglang/srt/entrypoints/openai/serving_chat.py @@ -51,7 +51,7 @@ from sglang.srt.managers.io_struct import GenerateReqInput from sglang.srt.parser.conversation import generate_chat_conv from sglang.srt.parser.jinja_template_utils import process_content_for_template_format -from sglang.srt.parser.reasoning_parser import ReasoningParser +from sglang.srt.parser.reasoning_parser import ReasoningParser, StreamingParseResult if TYPE_CHECKING: from sglang.srt.managers.template_manager import TemplateManager @@ -59,6 +59,8 @@ logger = logging.getLogger(__name__) +_NORMAL_STREAM_FINISH_REASONS = frozenset({"stop", "length"}) + def _extract_max_dynamic_patch(request: ChatCompletionRequest): img_vals = [] @@ -128,7 +130,9 @@ def __init__( # markers. This also leaks structural specials (e.g. <|im_end|>) # into content; strip them post-parse to mirror # skip_special_tokens=True semantics. - self._special_token_strings: tuple[str, ...] = self._compute_special_token_strings() + self._special_token_strings: tuple[str, ...] = ( + self._compute_special_token_strings() + ) def _compute_special_token_strings(self) -> tuple[str, ...]: """Text form of every token marked special=True in the tokenizer. @@ -146,8 +150,10 @@ def _compute_special_token_strings(self) -> tuple[str, ...]: if isinstance(atd, dict): for _tid, info in atd.items(): # info may be a transformers AddedToken or a plain dict - is_special = bool(getattr(info, "special", None) or - (isinstance(info, dict) and info.get("special"))) + is_special = bool( + getattr(info, "special", None) + or (isinstance(info, dict) and info.get("special")) + ) if not is_special: continue content = getattr(info, "content", None) @@ -810,12 +816,41 @@ async def _generate_chat_stream( # Handle reasoning content if self.reasoning_parser and request.separate_reasoning: - reasoning_text, delta = self._process_reasoning_stream( + reasoning_result = self._process_reasoning_stream_result( index, delta, reasoning_parser_dict, content, request ) - reasoning_text = self._strip_special_tokens(reasoning_text) - delta = self._strip_special_tokens(delta) - if reasoning_text: + if finish_reason_type in _NORMAL_STREAM_FINISH_REASONS: + finalization = reasoning_parser_dict[ + index + ].finalize_reasoning_streaming() + if finalization is not None and finalization.result is not None: + final_result = finalization.result + if final_result.has_reasoning_text: + if reasoning_result.has_reasoning_text: + reasoning_result.reasoning_text += ( + final_result.reasoning_text + ) + else: + reasoning_result.reasoning_text = ( + final_result.reasoning_text + ) + reasoning_result.has_reasoning_text = True + if final_result.has_normal_text: + if reasoning_result.has_normal_text: + reasoning_result.normal_text += ( + final_result.normal_text + ) + else: + reasoning_result.normal_text = ( + final_result.normal_text + ) + reasoning_result.has_normal_text = True + + reasoning_text = self._strip_special_tokens( + reasoning_result.reasoning_text + ) + delta = self._strip_special_tokens(reasoning_result.normal_text) + if reasoning_result.has_reasoning_text: choice_data = ChatCompletionResponseStreamChoice( index=index, delta=DeltaMessage(reasoning_content=reasoning_text), @@ -859,6 +894,25 @@ async def _generate_chat_stream( # Send any remaining tool call arguments when generation finishes if finish_reason_type is not None and index in parser_dict: parser = parser_dict[index] + if finish_reason_type in _NORMAL_STREAM_FINISH_REASONS: + if isinstance(parser, FunctionCallParser): + normal_text, calls = parser.finalize_streaming() + else: + final_result = parser.finalize_streaming(request.tools) + normal_text, calls = ( + final_result.normal_text, + final_result.calls, + ) + async for chunk in self._emit_tool_call_stream_result( + index, + normal_text, + calls, + content, + request, + has_tool_calls, + continuous_usage_stats, + ): + yield chunk remaining_chunk = self._check_for_unstreamed_tool_args( parser, content, request, index ) @@ -897,9 +951,10 @@ async def _generate_chat_stream( for idx, finish_reason_data in finish_reasons.items(): finish_reason_type = finish_reason_data["type"] - # Change finish_reason to "tool_calls" if we had tool calls and stopped naturally + # Auto tool calls take precedence over the engine's terminal + # reason, including a length stop after a complete call. final_finish_reason = finish_reason_type - if has_tool_calls.get(idx, False) and finish_reason_type == "stop": + if has_tool_calls.get(idx, False): final_finish_reason = "tool_calls" finish_reason_chunk = ChatCompletionStreamResponse( @@ -1285,9 +1340,6 @@ def _process_tool_calls( chat_template_kwargs=chat_template_kwargs, ) if parser.has_tool_call(text): - if finish_reason["type"] == "stop": - finish_reason["type"] = "tool_calls" - finish_reason["matched"] = None try: text, call_info_list = parser.parse_non_stream(text) tool_calls = [] @@ -1304,6 +1356,9 @@ def _process_tool_calls( ), ) ) + if tool_calls: + finish_reason["type"] = "tool_calls" + finish_reason["matched"] = None return ToolCallProcessingResult(tool_calls, text, finish_reason) except Exception as e: logger.error(f"Tool call parsing error: {e}") @@ -1340,6 +1395,32 @@ def _process_reasoning_stream( request: ChatCompletionRequest, ) -> tuple[Optional[str], str]: """Process reasoning content in streaming response""" + reasoning_parser = self._get_or_create_reasoning_parser( + index, reasoning_parser_dict, request + ) + return reasoning_parser.parse_stream_chunk(delta) + + def _process_reasoning_stream_result( + self, + index: int, + delta: str, + reasoning_parser_dict: Dict[int, ReasoningParser], + content: Dict[str, Any], + request: ChatCompletionRequest, + ) -> StreamingParseResult: + """Process reasoning while preserving explicit empty boundary fields.""" + reasoning_parser = self._get_or_create_reasoning_parser( + index, reasoning_parser_dict, request + ) + return reasoning_parser.parse_stream_chunk_result(delta) + + def _get_or_create_reasoning_parser( + self, + index: int, + reasoning_parser_dict: Dict[int, ReasoningParser], + request: ChatCompletionRequest, + ) -> ReasoningParser: + """Return the per-choice reasoning parser, creating it on first use.""" if index not in reasoning_parser_dict: is_force_reasoning = ( self.template_manager.force_reasoning @@ -1351,8 +1432,7 @@ def _process_reasoning_stream( is_force_reasoning, request, ) - reasoning_parser = reasoning_parser_dict[index] - return reasoning_parser.parse_stream_chunk(delta) + return reasoning_parser_dict[index] def _get_history_tool_calls_cnt(self, request: ChatCompletionRequest) -> int: """Counts the number of tool calls in the request's message history. @@ -1458,6 +1538,29 @@ async def _process_tool_call_stream( else: normal_text, calls = parser.parse_stream_chunk(delta) + async for chunk in self._emit_tool_call_stream_result( + index, + normal_text, + calls, + content, + request, + has_tool_calls, + continuous_usage_stats, + ): + yield chunk + + async def _emit_tool_call_stream_result( + self, + index: int, + normal_text: Optional[str], + calls: List[ToolCallItem], + content: Dict[str, Any], + request: ChatCompletionRequest, + has_tool_calls: Dict[int, bool], + continuous_usage_stats: bool = False, + ): + """Serialize one parsed content/tool-call result as SSE chunks.""" + normal_text = self._strip_special_tokens(normal_text) # Yield normal text diff --git a/python/sglang/srt/entrypoints/openai/serving_responses.py b/python/sglang/srt/entrypoints/openai/serving_responses.py index 41aefac686cd..3c587db16c9b 100644 --- a/python/sglang/srt/entrypoints/openai/serving_responses.py +++ b/python/sglang/srt/entrypoints/openai/serving_responses.py @@ -57,7 +57,7 @@ from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat from sglang.srt.entrypoints.openai.tool_server import MCPToolServer, ToolServer from sglang.srt.managers.io_struct import GenerateReqInput -from sglang.srt.parser.reasoning_parser import ReasoningParser +from sglang.srt.parser.reasoning_parser import ReasoningParser, StreamingParseResult from sglang.srt.utils import random_uuid if TYPE_CHECKING: @@ -384,6 +384,9 @@ async def _make_request( model=request.model, messages=messages, stream=request.stream, + reasoning_effort=( + request.reasoning.effort if request.reasoning else None + ), ) # Follow SGLang's _process_messages pattern @@ -816,8 +819,8 @@ async def responses_stream_generator( self, request: ResponsesRequest, sampling_params: Any, - result_generator: AsyncIterator[StreamingHarmonyContext], - context: StreamingHarmonyContext, + result_generator: AsyncIterator[ConversationContext], + context: ConversationContext, model_name: str, tokenizer: Any, request_metadata: RequestResponseMetadata, @@ -872,6 +875,26 @@ def _send_event(event): ) ) + if not self.use_harmony: + async for event in self._process_simple_streaming_events( + request=request, + result_generator=result_generator, + ): + yield _send_event(event) + + yield _send_event( + await self._create_streaming_completed_event( + request=request, + sampling_params=sampling_params, + context=context, + model_name=model_name, + tokenizer=tokenizer, + request_metadata=request_metadata, + created_time=created_time, + ) + ) + return + async for ctx in result_generator: # Only process context objects that implement the `is_expecting_start()` method, @@ -1215,6 +1238,30 @@ def _send_event(event): ) ) + yield _send_event( + await self._create_streaming_completed_event( + request=request, + sampling_params=sampling_params, + context=context, + model_name=model_name, + tokenizer=tokenizer, + request_metadata=request_metadata, + created_time=created_time, + ) + ) + + async def _create_streaming_completed_event( + self, + request: ResponsesRequest, + sampling_params: Any, + context: ConversationContext, + model_name: str, + tokenizer: Any, + request_metadata: RequestResponseMetadata, + created_time: int, + ) -> openai_responses_types.ResponseCompletedEvent: + """Build the terminal event after the streaming generator is consumed.""" + async def empty_async_generator(): if False: yield @@ -1229,10 +1276,7 @@ async def empty_async_generator(): request_metadata, created_time=created_time, ) - # Convert final_response to the format expected by ResponseCompletedEvent response_dict = final_response.model_dump() - - # Convert UsageInfo to ResponseUsage format if response_dict.get("usage"): usage_info = response_dict["usage"] response_dict["usage"] = { @@ -1247,13 +1291,211 @@ async def empty_async_generator(): "total_tokens": usage_info.get("total_tokens", 0), } - yield _send_event( - openai_responses_types.ResponseCompletedEvent( - type="response.completed", - sequence_number=-1, - response=response_dict, + return openai_responses_types.ResponseCompletedEvent( + type="response.completed", + sequence_number=-1, + response=response_dict, + ) + + async def _process_simple_streaming_events( + self, + request: ResponsesRequest, + result_generator: AsyncIterator[ConversationContext], + ): + """Stream non-Harmony reasoning/content with terminal parser finalization.""" + reasoning_parser = ( + ReasoningParser( + model_type=self.reasoning_parser, + stream_reasoning=True, + request=request, ) + if self.reasoning_parser + else None ) + previous_text = "" + current_output_index = 0 + current_content_index = 0 + current_item_id = "" + current_kind: Optional[str] = None + current_text_parts: list[str] = [] + + def start_item(kind: str): + nonlocal current_item_id, current_kind, current_content_index + current_item_id = f"item_{random_uuid()}" + current_kind = kind + current_content_index = 0 + current_text_parts.clear() + if kind == "reasoning": + item = ResponseReasoningItem( + type="reasoning", + id=current_item_id, + summary=[], + status="in_progress", + ) + else: + item = openai_responses_types.ResponseOutputMessage( + id=current_item_id, + type="message", + role="assistant", + content=[], + status="in_progress", + ) + return [ + openai_responses_types.ResponseOutputItemAddedEvent( + type="response.output_item.added", + sequence_number=-1, + output_index=current_output_index, + item=item, + ), + openai_responses_types.ResponseContentPartAddedEvent( + type="response.content_part.added", + sequence_number=-1, + output_index=current_output_index, + item_id=current_item_id, + content_index=current_content_index, + part=openai_responses_types.ResponseOutputText( + type="output_text", + text="", + annotations=[], + logprobs=None, + ), + ), + ] + + def finish_item(): + if current_kind is None: + return [] + text = "".join(current_text_parts) + if current_kind == "reasoning": + item = ResponseReasoningItem( + type="reasoning", + content=[ + ResponseReasoningTextContent( + text=text, + type="reasoning_text", + ) + ], + status="completed", + id=current_item_id, + summary=[], + ) + return [ + openai_responses_types.ResponseReasoningTextDoneEvent( + type="response.reasoning_text.done", + item_id=current_item_id, + sequence_number=-1, + output_index=current_output_index, + content_index=current_content_index, + text=text, + ), + openai_responses_types.ResponseOutputItemDoneEvent( + type="response.output_item.done", + sequence_number=-1, + output_index=current_output_index, + item=item, + ), + ] + + part = openai_responses_types.ResponseOutputText( + text=text, + type="output_text", + annotations=[], + logprobs=None, + ) + item = openai_responses_types.ResponseOutputMessage( + type="message", + role="assistant", + content=[part], + status="completed", + id=current_item_id, + ) + return [ + openai_responses_types.ResponseTextDoneEvent( + type="response.output_text.done", + sequence_number=-1, + output_index=current_output_index, + content_index=current_content_index, + text=text, + logprobs=[], + item_id=current_item_id, + ), + openai_responses_types.ResponseContentPartDoneEvent( + type="response.content_part.done", + sequence_number=-1, + item_id=current_item_id, + output_index=current_output_index, + content_index=current_content_index, + part=part, + ), + openai_responses_types.ResponseOutputItemDoneEvent( + type="response.output_item.done", + sequence_number=-1, + output_index=current_output_index, + item=item, + ), + ] + + async for ctx in result_generator: + assert isinstance(ctx, SimpleContext) + output = ctx.last_output + if output is None: + continue + current_text = output["text"] + delta_text = current_text[len(previous_text) :] + previous_text = current_text + finish_reason = output.get("meta_info", {}).get("finish_reason") + finish_reason_type = finish_reason.get("type") if finish_reason else None + + results = [] + if reasoning_parser is not None: + results.append(reasoning_parser.parse_stream_chunk_result(delta_text)) + if finish_reason_type in {"stop", "length"}: + finalization = reasoning_parser.finalize_reasoning_streaming() + if finalization is not None and finalization.result is not None: + results.append(finalization.result) + else: + results.append(StreamingParseResult(normal_text=delta_text)) + + for result in results: + visible_deltas = [] + if result.has_reasoning_text and result.reasoning_text: + visible_deltas.append(("reasoning", result.reasoning_text)) + if result.has_normal_text: + visible_deltas.append(("content", result.normal_text)) + + for kind, text_delta in visible_deltas: + if current_kind != kind: + if current_kind is not None: + for event in finish_item(): + yield event + current_output_index += 1 + for event in start_item(kind): + yield event + + current_text_parts.append(text_delta) + if kind == "reasoning": + event = openai_responses_types.ResponseReasoningTextDeltaEvent( + type="response.reasoning_text.delta", + item_id=current_item_id, + output_index=current_output_index, + content_index=current_content_index, + delta=text_delta, + sequence_number=-1, + ) + else: + event = openai_responses_types.ResponseTextDeltaEvent( + type="response.output_text.delta", + sequence_number=-1, + content_index=current_content_index, + output_index=current_output_index, + item_id=current_item_id, + delta=text_delta, + logprobs=[], + ) + yield event + + for event in finish_item(): + yield event async def _generate_with_builtin_tools( self, diff --git a/python/sglang/srt/function_call/base_format_detector.py b/python/sglang/srt/function_call/base_format_detector.py index 3163867bcd05..3d7eb68ebaa6 100644 --- a/python/sglang/srt/function_call/base_format_detector.py +++ b/python/sglang/srt/function_call/base_format_detector.py @@ -270,7 +270,7 @@ def parse_streaming_increment( cur_arguments = current_tool_call.get("arguments") res = StreamingParseResult() - if cur_arguments: + if cur_arguments is not None: # Calculate how much of the arguments we've already streamed sent = len(self.streamed_args_for_tool[self.current_tool_id]) cur_args_json = json.dumps(cur_arguments, ensure_ascii=False) @@ -344,6 +344,14 @@ def has_tool_call(self, text: str) -> bool: """ raise NotImplementedError() + def finalize_streaming(self, tools: List[Tool]) -> StreamingParseResult: + """Release text held by the parser after normal model completion.""" + return StreamingParseResult() + + def has_pending_streaming_output(self) -> bool: + """Return whether source text is held without a complete output delta.""" + return False + def supports_structural_tag(self) -> bool: """Return True if this detector supports structural tag format.""" return True diff --git a/python/sglang/srt/function_call/function_call_parser.py b/python/sglang/srt/function_call/function_call_parser.py index 2bdd4010966e..56748371acf3 100644 --- a/python/sglang/srt/function_call/function_call_parser.py +++ b/python/sglang/srt/function_call/function_call_parser.py @@ -165,6 +165,17 @@ def parse_stream_chunk(self, chunk_text: str) -> Tuple[str, list[ToolCallItem]]: return final_normal_text, final_calls + def finalize_streaming(self) -> Tuple[str, list[ToolCallItem]]: + """Release detector-held text after normal model completion.""" + if not self.tools: + return "", [] + result = self.detector.finalize_streaming(self.tools) + return result.normal_text or "", result.calls + + def has_pending_streaming_output(self) -> bool: + """Return whether the detector is holding unresolved source text.""" + return self.detector.has_pending_streaming_output() + def get_structure_tag(self) -> LegacyStructuralTagResponseFormat: """ Generate a structural tag response format for all available tools. diff --git a/python/sglang/srt/function_call/json_array_parser.py b/python/sglang/srt/function_call/json_array_parser.py index e38e3b1b0563..9209b19a9a17 100644 --- a/python/sglang/srt/function_call/json_array_parser.py +++ b/python/sglang/srt/function_call/json_array_parser.py @@ -42,6 +42,38 @@ def parse_streaming_increment( """ return super().parse_streaming_increment(new_text, tools) + def finalize_streaming(self, tools: List[Tool]) -> StreamingParseResult: + """Drain complete JSON calls still buffered in the terminal engine delta.""" + normal_text = "" + calls = [] + + # BaseFormatDetector emits a call's name and arguments on separate + # invocations. A terminal engine delta can contain the entire JSON + # array, so keep parsing the existing buffer until no state changes. + max_steps = 2 * (self._buffer.count(self.tool_call_separator) + 1) + 2 + for _ in range(max_steps): + before = ( + self._buffer, + self.current_tool_id, + self.current_tool_name_sent, + tuple(self.streamed_args_for_tool), + repr(self.prev_tool_call_arr), + ) + result = super().parse_streaming_increment("", tools) + normal_text += result.normal_text or "" + calls.extend(result.calls) + after = ( + self._buffer, + self.current_tool_id, + self.current_tool_name_sent, + tuple(self.streamed_args_for_tool), + repr(self.prev_tool_call_arr), + ) + if after == before: + break + + return StreamingParseResult(normal_text=normal_text, calls=calls) + def structure_info(self) -> callable: """ Return a function that creates StructureInfo for constrained generation. diff --git a/python/sglang/srt/function_call/multi_format_detector.py b/python/sglang/srt/function_call/multi_format_detector.py index 14a540a4706e..86248bfbe10b 100644 --- a/python/sglang/srt/function_call/multi_format_detector.py +++ b/python/sglang/srt/function_call/multi_format_detector.py @@ -1,6 +1,6 @@ """Multi-format tool-call detector that dispatches on a per-request tool_format. -Ported from vLLM's MultiFormatToolParser (v0.12.0-ifm_xllm-fix branch). +Ported from LLM360/vllm's MultiFormatToolParser, including PR #12. Dialects: - "default" : delegate to HermesDetector @@ -16,16 +16,10 @@ Streaming: - "default"/"qwen3" delegate to their sub-detectors (which stream natively). - - The IFM dialects ("xml", "xml_typed", "json") — the K2-V3 format — DO stream: - * xml / xml_typed: a character-by-character XML->JSON state machine - (modeled on sglang's Glm4Moe/Glm47Moe detectors) emits the tool name - first, then incremental JSON argument fragments. - * json: streams at tool-call-block granularity (each - block is emitted as soon as it completes). - vLLM's K2V3ToolParser does NOT stream the IFM format; this is a deliberate - SGLang enhancement and the reassembled deltas match the non-stream parse. - - The remaining embedded dialects (minimax, dsv32, glm, gptoss, python) still - buffer and emit nothing during streaming (parse runs on the final call). + - Embedded dialects buffer from the first possible tool marker and repeatedly + use the non-streaming parser. K2-V3 further requires a complete plural + wrapper before emitting calls. On normal completion, + unresolved markup is released as ordinary content. """ from __future__ import annotations @@ -34,7 +28,6 @@ import json import logging import re -from enum import Enum from typing import Any, List, Optional from sglang.srt.entrypoints.openai.protocol import Tool @@ -47,21 +40,6 @@ logger = logging.getLogger(__name__) - -class _IfmStreamState(str, Enum): - """State machine states for the IFM XML->JSON streaming converter. - - Mirrors the GLM detectors' StreamState, with an extra IN_TYPE state for the - optional ```` hint that sits between key and value. - """ - - INIT = "INIT" - BETWEEN = "BETWEEN" - IN_KEY = "IN_KEY" - AFTER_KEY = "AFTER_KEY" # waiting for or - IN_TYPE = "IN_TYPE" - IN_VALUE = "IN_VALUE" - _EMBEDDED_DIALECTS = { "minimax", "dsv32", @@ -110,11 +88,9 @@ def __init__( self._delegate = Qwen3CoderDetector() - # Streaming state for the IFM dialects (xml/xml_typed/json). Unused by - # the delegating dialects but harmless to initialize. - self._last_arguments = "" - self._streamed_raw_length = 0 - self._reset_ifm_stream_state() + self._streaming_tool_call_started = False + self._streaming_content_buffer = "" + self._streaming_tool_calls_emitted = 0 # BaseFormatDetector contract ------------------------------------- @@ -133,9 +109,7 @@ def has_tool_call(self, text: str) -> bool: return "" in text and "to=functions." in text return False - def detect_and_parse( - self, text: str, tools: List[Tool] - ) -> StreamingParseResult: + def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult: if self._delegate is not None: return self._delegate.detect_and_parse(text, tools) @@ -170,515 +144,110 @@ def parse_streaming_increment( ) -> StreamingParseResult: if self._delegate is not None: return self._delegate.parse_streaming_increment(new_text, tools) - if self.tool_format in ("xml", "xml_typed"): - return self._ifm_xml_streaming_increment(new_text, tools) - if self.tool_format == "json": - return self._ifm_json_streaming_increment(new_text, tools) - # Remaining embedded dialects (minimax/dsv32/glm/gptoss/python) do not - # stream: buffer the text and emit nothing. detect_and_parse runs on the - # complete text passed by the serving layer, so this buffer is unused. - self._buffer += new_text - return StreamingParseResult() - - # IFM streaming (K2-V3 format) ------------------------------------ - # - # vLLM's K2V3ToolParser does not stream the IFM dialects; the logic below is - # an SGLang enhancement modeled on Glm47MoeDetector (whose - # ``name....`` format is structurally - # identical to ``name....``), - # extended for the optional ```` hint. - - # (Tool-call open/close token constants live with the other IFM tokens in - # the "Class-level regex constants for IFM" block below.) - _IFM_ARG_KEY_OPEN = "" - _IFM_ARG_KEY_OPEN_PREFIX = "" - _IFM_ARG_KEY_CLOSE = "" - _IFM_ARG_TYPE_OPEN = "" - _IFM_ARG_TYPE_CLOSE = "" - _IFM_ARG_VALUE_OPEN = "" - _IFM_ARG_VALUE_CLOSE = "" - - _IFM_STREAM_TOOL_CALL_REGEX = re.compile( - r"(.*?)(?:()|$)", - re.DOTALL, - ) + return self._streaming_increment_fallback(new_text, tools) - def _reset_ifm_stream_state(self) -> None: - """Reset the per-tool-call IFM streaming state machine.""" - self._ifm_state = _IfmStreamState.INIT - self._ifm_current_key = "" - self._ifm_current_value = "" - self._ifm_tag_buffer = "" - self._ifm_is_first_param = True - self._ifm_value_started = False - self._ifm_cached_value_type: Optional[str] = None - self._ifm_inline_type: Optional[str] = None - self._ifm_tool_completed = False - self._ifm_sent_empty_object = False - - # Wrapper/closing tokens that frame tool calls and must never leak as - # content. (The opening "" is handled by the main branch.) - _IFM_SUPPRESS_TOKENS = ( - "", - "", - "", - ) - # Proper prefixes of an opening " Optional[int]: - """Earliest index at which an IFM marker (opening `` tuple[str, str]: - """Split a no-tool-call buffer into (emit, hold). - - Genuine content (including whitespace) is emitted; any IFM markup — a - leading reasoning block, a tool-call wrapper/closing token, or a partial - `` Optional[str]: - """Normal text from the segment before the first ```` in - the same chunk: reasoning/wrapper stripped while preserving genuine - whitespace content.""" - emit, _ = self._ifm_split_normal_text(prefix) - return emit if emit != "" else None - - def _ifm_stream_value_type( - self, func_name: str, key: str, tools: List[Tool] - ) -> str: - """Resolve the streaming value type for an argument. - - Precedence matches the non-stream ``_coerce_argument_value``: schema type - then the inline ```` hint. Untyped values default to - "string" (a streaming limitation shared with the GLM detectors, since the - full value is not yet known when its type must be decided).""" - target = self._schema_arg_type(func_name, key, tools) or self._ifm_inline_type - if self._arg_type_preserves_text(target): - return "string" - if target in ("number", "integer", "float"): - return "number" - if target is not None: - # object / array / boolean: stream the model's verbatim JSON. - return "raw" - return "string" - - def _process_ifm_xml_to_json_streaming( - self, raw_increment: str, func_name: str, tools: List[Tool] - ) -> str: - """Convert an IFM XML increment to a JSON increment, char by char, - preserving state across calls to handle tags/values split across chunks.""" - json_output = "" - - for char in raw_increment: - self._ifm_tag_buffer += char - - if self._ifm_state in (_IfmStreamState.INIT, _IfmStreamState.BETWEEN): - if self._ifm_tag_buffer.endswith(self._IFM_ARG_KEY_OPEN): - self._ifm_state = _IfmStreamState.IN_KEY - self._ifm_current_key = "" - self._ifm_tag_buffer = "" - json_output += "{" if self._ifm_is_first_param else ", " - self._ifm_is_first_param = False - - elif self._ifm_state == _IfmStreamState.IN_KEY: - if self._ifm_tag_buffer.endswith(self._IFM_ARG_KEY_CLOSE): - self._ifm_current_key = self._ifm_tag_buffer[ - : -len(self._IFM_ARG_KEY_CLOSE) - ].strip() - self._ifm_tag_buffer = "" - self._ifm_state = _IfmStreamState.AFTER_KEY - json_output += ( - json.dumps(self._ifm_current_key, ensure_ascii=False) + ": " - ) - - elif self._ifm_state == _IfmStreamState.AFTER_KEY: - if self._ifm_tag_buffer.endswith(self._IFM_ARG_TYPE_OPEN): - self._ifm_state = _IfmStreamState.IN_TYPE - self._ifm_tag_buffer = "" - elif self._ifm_tag_buffer.endswith(self._IFM_ARG_VALUE_OPEN): - self._ifm_state = _IfmStreamState.IN_VALUE - self._ifm_current_value = "" - self._ifm_tag_buffer = "" - self._ifm_value_started = False - self._ifm_cached_value_type = self._ifm_stream_value_type( - func_name, self._ifm_current_key, tools - ) + def finalize_streaming(self, tools: List[Tool]) -> StreamingParseResult: + if self._delegate is not None: + return self._delegate.finalize_streaming(tools) + if not self._streaming_content_buffer: + return StreamingParseResult() + if self._streaming_tool_calls_emitted: + return StreamingParseResult() + content = self._streaming_content_buffer + self._streaming_content_buffer = "" + return StreamingParseResult(normal_text=content) + + def has_pending_streaming_output(self) -> bool: + if self._delegate is not None: + return self._delegate.has_pending_streaming_output() + return bool( + self._streaming_content_buffer and not self._streaming_tool_calls_emitted + ) - elif self._ifm_state == _IfmStreamState.IN_TYPE: - if self._ifm_tag_buffer.endswith(self._IFM_ARG_TYPE_CLOSE): - self._ifm_inline_type = self._ifm_tag_buffer[ - : -len(self._IFM_ARG_TYPE_CLOSE) - ].strip() or None - self._ifm_tag_buffer = "" - self._ifm_state = _IfmStreamState.AFTER_KEY - - elif self._ifm_state == _IfmStreamState.IN_VALUE: - if self._ifm_tag_buffer.endswith(self._IFM_ARG_VALUE_CLOSE): - final_value = self._ifm_tag_buffer[ - : -len(self._IFM_ARG_VALUE_CLOSE) - ] - self._ifm_current_value += final_value - value_type = self._ifm_cached_value_type or "string" - - if self._ifm_value_started: - if final_value: - if value_type == "string": - json_output += json.dumps( - final_value, ensure_ascii=False - )[1:-1] - else: - json_output += final_value - if value_type == "string": - json_output += '"' - else: - json_output += self._ifm_format_value_complete( - self._ifm_current_value, value_type - ) - - self._ifm_tag_buffer = "" - self._ifm_state = _IfmStreamState.BETWEEN - self._ifm_current_value = "" - self._ifm_value_started = False - self._ifm_cached_value_type = None - self._ifm_inline_type = None - else: - closing = self._IFM_ARG_VALUE_CLOSE - is_potential_closing = len(self._ifm_tag_buffer) <= len( - closing - ) and closing.startswith(self._ifm_tag_buffer) - if not is_potential_closing: - content = self._ifm_tag_buffer - value_type = self._ifm_cached_value_type or "string" - if value_type == "string": - if not self._ifm_value_started: - json_output += '"' - self._ifm_value_started = True - if content: - json_output += json.dumps(content, ensure_ascii=False)[ - 1:-1 - ] - self._ifm_current_value += content - self._ifm_tag_buffer = "" - else: - if content: - if not self._ifm_value_started: - self._ifm_value_started = True - json_output += content - self._ifm_current_value += content - self._ifm_tag_buffer = "" - - return json_output + def _tool_call_markers(self) -> tuple[str, ...]: + if self.tool_format in ("json", "xml", "xml_typed"): + return ( + self._IFM_TOOL_CALLS_START_TOKEN, + self._IFM_TOOL_CALL_START_TOKEN, + ) + if self.tool_format in ("minimax", "dsv32"): + return (self._MINIMAX_START,) + if self.tool_format == "glm": + return ( + self._IFM_TOOL_CALLS_START_TOKEN, + self._IFM_TOOL_CALL_START_TOKEN, + "", + ) + return ("",) + + def _partial_tool_call_marker_start(self, text: str) -> Optional[int]: + markers = self._tool_call_markers() + max_marker_len = max(len(marker) for marker in markers) + start = max(0, len(text) - max_marker_len + 1) + for index in range(start, len(text)): + suffix = text[index:] + if any(marker.startswith(suffix) for marker in markers): + return index + return None - @staticmethod - def _ifm_format_value_complete(value: str, value_type: str) -> str: - """Format a value that arrived in a single chunk (state machine never - emitted an opening quote for it).""" - if value_type == "string": - return json.dumps(value, ensure_ascii=False) - if value_type == "number": - stripped = value.strip() - try: - if "." in stripped or "e" in stripped.lower(): - return str(float(stripped)) - return str(int(stripped)) - except (ValueError, AttributeError): - return json.dumps(value, ensure_ascii=False) - # object / array / boolean: already valid JSON. Guard an empty value - # (malformed input) that would otherwise emit nothing -> invalid JSON. - return value if value else '""' - - def _ifm_xml_streaming_increment( - self, new_text: str, tools: List[Tool] + def _try_emit_streaming_tool_calls( + self, tools: List[Tool], content: Optional[str] = None ) -> StreamingParseResult: - self._buffer += new_text - current_text = self._buffer - start_token = self._IFM_TOOL_CALL_START_TOKEN - - normal_text = "" - calls: List[ToolCallItem] = [] - - if not hasattr(self, "_tool_indices"): - self._tool_indices = self._get_tool_indices(tools) - - try: - while current_text: - if start_token not in current_text: - if calls: - self._buffer = current_text - break - emit, hold = self._ifm_split_normal_text(current_text) - normal_text += emit - self._buffer = hold - break - - first_idx = current_text.find(start_token) - if first_idx > 0: - if calls: - self._buffer = current_text - break - prefix_normal_text = self._ifm_prefix_normal_text( - current_text[:first_idx] - ) - if prefix_normal_text is not None: - normal_text += prefix_normal_text - current_text = current_text[first_idx:] - - partial_match = self._IFM_STREAM_TOOL_CALL_REGEX.search(current_text) - if not partial_match: - self._buffer = current_text - break - - func_name = partial_match.group(1).strip() - func_args_raw = ( - partial_match.group(2).strip() if partial_match.group(2) else "" - ) - is_tool_end = partial_match.group(3) or "" - - if self.current_tool_id == -1: - self.current_tool_id = 0 - self.prev_tool_call_arr = [] - self.streamed_args_for_tool = [""] - self._streamed_raw_length = 0 - self.current_tool_name_sent = False - self._reset_ifm_stream_state() - - while len(self.prev_tool_call_arr) <= self.current_tool_id: - self.prev_tool_call_arr.append({}) - while len(self.streamed_args_for_tool) <= self.current_tool_id: - self.streamed_args_for_tool.append("") - - matched_text = current_text[: partial_match.end()] - has_arg_key = self._IFM_ARG_KEY_OPEN_PREFIX in matched_text - - name_item = self._ifm_send_tool_name( - func_name, has_arg_key, is_tool_end - ) - if name_item: - calls.append(name_item) + if not self._streaming_input_complete(): + return StreamingParseResult(normal_text=content, calls=[]) - if self.current_tool_name_sent: - arg_item = self._ifm_process_arguments( - func_name, func_args_raw, tools - ) - if arg_item: - calls.append(arg_item) - - if ( - is_tool_end == self._IFM_TOOL_CALL_END_TOKEN - and not self._ifm_tool_completed - ): - calls.extend( - self._finalize_ifm_xml_tool_call( - func_name, func_args_raw, tools - ) - ) - current_text = current_text[partial_match.end() :] - self._buffer = current_text - self.current_tool_id += 1 - self._last_arguments = "" - self.current_tool_name_sent = False - self._streamed_raw_length = 0 - self._reset_ifm_stream_state() - continue - - self._buffer = current_text - break - - except Exception: - logger.exception( - "MultiFormatDetector IFM xml streaming failed for tool_format=%s", - self.tool_format, - ) - return StreamingParseResult(normal_text=current_text) - - return StreamingParseResult(normal_text=normal_text, calls=calls) - - def _ifm_send_tool_name( - self, func_name: str, has_arg_key: bool, is_tool_end: str - ) -> Optional[ToolCallItem]: - """Emit the tool name once it is known — an ```` or the - closing ```` proves the name is complete. Returns the - name item (with empty parameters), or None if the name is not yet ready - or was already sent.""" - if self.current_tool_name_sent: - return None - name_complete = has_arg_key or is_tool_end == self._IFM_TOOL_CALL_END_TOKEN - if not (name_complete and func_name): - return None - self.current_tool_name_sent = True - self._streamed_raw_length = 0 - self._reset_ifm_stream_state() - self.prev_tool_call_arr[self.current_tool_id] = { - "name": func_name, - "arguments": {}, - } - return ToolCallItem( - tool_index=self.current_tool_id, name=func_name, parameters="" - ) + parsed = self.detect_and_parse(self._streaming_content_buffer, tools) + if not parsed.calls: + return StreamingParseResult(normal_text=content, calls=[]) - def _ifm_process_arguments( - self, func_name: str, func_args_raw: str, tools: List[Tool] - ) -> Optional[ToolCallItem]: - """Feed newly-arrived raw argument text through the XML->JSON state - machine and return the JSON increment as a ToolCallItem, or None when - nothing new was produced this chunk.""" - current_raw_length = len(func_args_raw) - if current_raw_length <= self._streamed_raw_length: - return None - raw_increment = func_args_raw[self._streamed_raw_length :] - json_increment = self._process_ifm_xml_to_json_streaming( - raw_increment, func_name, tools - ) - # Advance even when no JSON was produced: the input has been consumed by - # the state machine (it may be buffering a partial tag). - self._streamed_raw_length = current_raw_length - if not json_increment: - return None - self._last_arguments += json_increment - self.streamed_args_for_tool[self.current_tool_id] += json_increment - return ToolCallItem( - tool_index=self.current_tool_id, name=None, parameters=json_increment - ) + new_calls = parsed.calls[self._streaming_tool_calls_emitted :] + if not new_calls: + return StreamingParseResult(normal_text=content, calls=[]) - def _finalize_ifm_xml_tool_call( - self, func_name: str, func_args_raw: str, tools: List[Tool] - ) -> List[ToolCallItem]: - """Close out the current tool call: emit the closing brace (or {} for a - no-arg call) and record the fully-parsed arguments for end-of-stream - flushing by the serving layer.""" - calls: List[ToolCallItem] = [] + for call in new_calls: + call.tool_index = self._streaming_tool_calls_emitted - if self._ifm_is_first_param and not self._ifm_sent_empty_object: - calls.append( - ToolCallItem( - tool_index=self.current_tool_id, name=None, parameters="{}" - ) - ) - self._last_arguments += "{}" - self.streamed_args_for_tool[self.current_tool_id] += "{}" - self._ifm_sent_empty_object = True - elif not self._last_arguments.endswith("}") and not self._ifm_sent_empty_object: - calls.append( - ToolCallItem( - tool_index=self.current_tool_id, name=None, parameters="}" - ) - ) - self._last_arguments += "}" - self.streamed_args_for_tool[self.current_tool_id] += "}" - self._ifm_sent_empty_object = True + try: + arguments = json.loads(call.parameters) + except (TypeError, json.JSONDecodeError): + arguments = call.parameters + self.prev_tool_call_arr.append({"name": call.name, "arguments": arguments}) + self.streamed_args_for_tool.append(call.parameters) + self._streaming_tool_calls_emitted += 1 - # Record the final parsed arguments (reuse the non-stream extractor on the - # reconstructed block) so the serving layer can flush any unstreamed - # remainder consistently. - try: - block = func_name + func_args_raw - for _, args in self._ifm_xml_calls(block, tools): - self.prev_tool_call_arr[self.current_tool_id]["arguments"] = args - except Exception: - logger.debug("IFM xml finalize: argument re-parse failed", exc_info=True) + return StreamingParseResult(normal_text=content, calls=new_calls) - self._ifm_tool_completed = True - return calls + def _streaming_input_complete(self) -> bool: + return True - def _ifm_json_streaming_increment( + def _streaming_increment_fallback( self, new_text: str, tools: List[Tool] ) -> StreamingParseResult: - """Stream the IFM ``json`` dialect at tool-call-block granularity: each - ``{json}`` block is parsed and emitted as - soon as it completes (name first, then the full argument JSON). A block - may itself contain a JSON list of calls, each emitted in turn.""" - self._buffer += new_text - current_text = self._buffer - start_token = self._IFM_TOOL_CALL_START_TOKEN - - if start_token not in current_text: - emit, hold = self._ifm_split_normal_text(current_text) - self._buffer = hold - return StreamingParseResult(normal_text=emit) - - if not hasattr(self, "_tool_indices"): - self._tool_indices = self._get_tool_indices(tools) - - normal_text = "" - first_idx = current_text.find(start_token) - if first_idx > 0: - prefix_normal_text = self._ifm_prefix_normal_text(current_text[:first_idx]) - if prefix_normal_text is not None: - normal_text = prefix_normal_text - current_text = current_text[first_idx:] - - if self.current_tool_id == -1: - self.current_tool_id = 0 - self.prev_tool_call_arr = [] - self.streamed_args_for_tool = [""] + self._streaming_content_buffer += new_text - calls: List[ToolCallItem] = [] - consumed = 0 - for match in self._IFM_BLOCK_REGEX.finditer(current_text): - try: - for name, args in self._ifm_json_calls(match.group(1), tools): - args_json = json.dumps(args, ensure_ascii=False) - while len(self.prev_tool_call_arr) <= self.current_tool_id: - self.prev_tool_call_arr.append({}) - while len(self.streamed_args_for_tool) <= self.current_tool_id: - self.streamed_args_for_tool.append("") - self.prev_tool_call_arr[self.current_tool_id] = { - "name": name, - "arguments": args, - } - calls.append( - ToolCallItem( - tool_index=self.current_tool_id, - name=name, - parameters="", - ) - ) - calls.append( - ToolCallItem( - tool_index=self.current_tool_id, - name=None, - parameters=args_json, - ) - ) - self.streamed_args_for_tool[self.current_tool_id] = args_json - self.current_tool_id += 1 - except Exception: - logger.exception( - "MultiFormatDetector IFM json streaming failed for a block" - ) - consumed = match.end() + if self._streaming_tool_call_started: + return self._try_emit_streaming_tool_calls(tools) - self._buffer = current_text[consumed:] if consumed else current_text - return StreamingParseResult(normal_text=normal_text, calls=calls) + buffer = self._streaming_content_buffer + markers = self._tool_call_markers() + marker_positions = [ + position for marker in markers if (position := buffer.find(marker)) != -1 + ] + if marker_positions: + first_marker_position = min(marker_positions) + self._streaming_tool_call_started = True + content = buffer[:first_marker_position] + self._streaming_content_buffer = buffer[first_marker_position:] + return self._try_emit_streaming_tool_calls(tools, content or None) + + partial_start = self._partial_tool_call_marker_start(buffer) + if partial_start is not None: + content = buffer[:partial_start] + self._streaming_content_buffer = buffer[partial_start:] + return StreamingParseResult(normal_text=content or None, calls=[]) + + self._streaming_content_buffer = "" + return StreamingParseResult(normal_text=buffer or None, calls=[]) def supports_structural_tag(self) -> bool: if self._delegate is not None: @@ -1007,7 +576,7 @@ def _coerce_arguments( r".*?|" r".*?|" r".*?" - r")", + r")\s*", re.DOTALL, ) @@ -1149,3 +718,61 @@ def __init__( if tool_format is None: tool_format = chat_template_kwargs.get("tool_call_format") super().__init__(tool_format=tool_format or "xml") + + def has_tool_call(self, text: str) -> bool: + uses_ifm_format = self.tool_format in ("json", "xml", "xml_typed") + contains_ifm_marker = ( + self._IFM_TOOL_CALLS_START_TOKEN in text + or self._IFM_TOOL_CALL_START_TOKEN in text + ) + if uses_ifm_format or (self.tool_format == "glm" and contains_ifm_marker): + wrapper_start = text.find(self._IFM_TOOL_CALLS_START_TOKEN) + if wrapper_start == -1: + return False + return ( + text.find( + self._IFM_TOOL_CALLS_END_TOKEN, + wrapper_start + len(self._IFM_TOOL_CALLS_START_TOKEN), + ) + != -1 + ) + return super().has_tool_call(text) + + def _tool_call_markers(self) -> tuple[str, ...]: + if self.tool_format in ("json", "xml", "xml_typed"): + return (self._IFM_TOOL_CALLS_START_TOKEN,) + if self.tool_format == "glm": + return (self._IFM_TOOL_CALLS_START_TOKEN, "") + return super()._tool_call_markers() + + def _streaming_input_complete(self) -> bool: + if self._IFM_TOOL_CALLS_START_TOKEN in self._streaming_content_buffer: + return self._IFM_TOOL_CALLS_END_TOKEN in self._streaming_content_buffer + return super()._streaming_input_complete() + + def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult: + uses_ifm_format = self.tool_format in ("json", "xml", "xml_typed") + contains_ifm_marker = ( + self._IFM_TOOL_CALLS_START_TOKEN in text + or self._IFM_TOOL_CALL_START_TOKEN in text + ) + if uses_ifm_format or (self.tool_format == "glm" and contains_ifm_marker): + wrapper_start = text.find(self._IFM_TOOL_CALLS_START_TOKEN) + wrapper_end = text.find( + self._IFM_TOOL_CALLS_END_TOKEN, + wrapper_start + len(self._IFM_TOOL_CALLS_START_TOKEN), + ) + if wrapper_start == -1 or wrapper_end == -1: + return StreamingParseResult(normal_text=text, calls=[]) + wrapper_end += len(self._IFM_TOOL_CALLS_END_TOKEN) + extracted = super().detect_and_parse(text[wrapper_start:wrapper_end], tools) + if not extracted.calls: + return StreamingParseResult(normal_text=text, calls=[]) + prefix = self._strip_ifm_reasoning_prefix(text[:wrapper_start]) + extracted.normal_text = prefix + return extracted + + extracted = super().detect_and_parse(text, tools) + if extracted.calls and extracted.normal_text is None: + extracted.normal_text = "" + return extracted diff --git a/python/sglang/srt/parser/reasoning_parser.py b/python/sglang/srt/parser/reasoning_parser.py index 5b9babefc0d7..3a771fdae811 100644 --- a/python/sglang/srt/parser/reasoning_parser.py +++ b/python/sglang/srt/parser/reasoning_parser.py @@ -1,19 +1,37 @@ -from typing import Dict, Optional, Tuple, Type +from dataclasses import dataclass +from typing import Dict, Optional, Tuple, Type, Union -from sglang.srt.entrypoints.openai.protocol import ChatCompletionRequest +from sglang.srt.entrypoints.openai.protocol import ( + ChatCompletionRequest, + ResponsesRequest, +) from sglang.srt.parser.harmony_parser import HarmonyParser +_UNSET = object() + class StreamingParseResult: """Result of streaming incremental parsing.""" def __init__( self, - normal_text: Optional[str] = None, - reasoning_text: Optional[str] = None, + normal_text: object = _UNSET, + reasoning_text: object = _UNSET, ): - self.normal_text = normal_text or "" - self.reasoning_text = reasoning_text or "" + self.has_normal_text = normal_text is not _UNSET + self.has_reasoning_text = reasoning_text is not _UNSET + self.normal_text = ( + "" if normal_text is _UNSET or normal_text is None else normal_text + ) + self.reasoning_text = ( + "" if reasoning_text is _UNSET or reasoning_text is None else reasoning_text + ) + + +@dataclass +class ReasoningParserStreamingFinalization: + result: Optional[StreamingParseResult] + reasoning_ended: bool class BaseReasoningFormatDetector: @@ -168,6 +186,11 @@ def parse_streaming_increment(self, new_text: str) -> StreamingParseResult: return StreamingParseResult() + def finalize_reasoning_streaming( + self, + ) -> Optional[ReasoningParserStreamingFinalization]: + return None + class DeepSeekR1Detector(BaseReasoningFormatDetector): """ @@ -488,7 +511,7 @@ class K2V3Detector(BaseReasoningFormatDetector): - medium: / - low: / - A tool call begins with ````; when the model emits one + A tool section begins with ````; when the model emits one without first closing the think block, reasoning is split at that boundary. The chat template inserts the start token into the prompt, so the @@ -506,9 +529,9 @@ class K2V3Detector(BaseReasoningFormatDetector): "low": ("", ""), } - # Boundary that ends reasoning when a tool call is emitted before the think - # end token. - _TOOL_START_TOKEN: str = "" + # Boundary that ends reasoning when a tool section is emitted before the + # think end token. + _TOOL_START_TOKEN: str = "" def __init__( self, @@ -535,6 +558,67 @@ def __init__( continue_final_message=continue_final_message, previous_content=previous_content, ) + self._streaming_state = "reasoning" if self._in_reasoning else "content" + self._streaming_reasoning_buffer = "" + + def _strip_optional_start_token(self, text: str) -> str: + if self.think_start_token in text: + _, _, text = text.partition(self.think_start_token) + return text + + def _split_model_output(self, text: str) -> tuple[str, str]: + text = self._strip_optional_start_token(text) + if self.think_end_token in text: + reasoning, _, content = text.partition(self.think_end_token) + return reasoning, content + if self._TOOL_START_TOKEN in text: + reasoning, marker, content = text.partition(self._TOOL_START_TOKEN) + return reasoning, marker + content + return "", text + + def detect_and_parse(self, text: str) -> StreamingParseResult: + reasoning, content = self._split_model_output(text) + return StreamingParseResult( + normal_text=content, + reasoning_text=reasoning, + ) + + def parse_streaming_increment(self, new_text: str) -> StreamingParseResult: + if self._streaming_state == "content": + return StreamingParseResult(normal_text=new_text) + + self._streaming_reasoning_buffer += new_text + if self.think_end_token not in self._streaming_reasoning_buffer: + return StreamingParseResult() + + reasoning, content = self._split_model_output(self._streaming_reasoning_buffer) + self._streaming_reasoning_buffer = "" + self._streaming_state = "content" + self._in_reasoning = False + if content: + return StreamingParseResult( + normal_text=content, + reasoning_text=reasoning, + ) + return StreamingParseResult(reasoning_text=reasoning) + + def finalize_reasoning_streaming( + self, + ) -> Optional[ReasoningParserStreamingFinalization]: + if self._streaming_state != "reasoning": + return None + + reasoning, content = self._split_model_output(self._streaming_reasoning_buffer) + self._streaming_reasoning_buffer = "" + self._streaming_state = "content" + self._in_reasoning = False + return ReasoningParserStreamingFinalization( + result=StreamingParseResult( + normal_text=content, + reasoning_text=reasoning, + ), + reasoning_ended=True, + ) class K2V3DetectorLegacy(K2V3Detector): @@ -562,6 +646,14 @@ class K2V3DetectorLegacy(K2V3Detector): _TOOL_START_TOKEN: str = "" + # Preserve the legacy detector's eager streaming behavior. The buffering + # and terminal classification above apply only to canonical K2 IFM output. + detect_and_parse = BaseReasoningFormatDetector.detect_and_parse + parse_streaming_increment = BaseReasoningFormatDetector.parse_streaming_increment + finalize_reasoning_streaming = ( + BaseReasoningFormatDetector.finalize_reasoning_streaming + ) + class ReasoningParser: """ @@ -600,7 +692,7 @@ def __init__( model_type: Optional[str] = None, stream_reasoning: bool = True, force_reasoning: Optional[bool] = None, - request: ChatCompletionRequest = None, + request: Optional[Union[ChatCompletionRequest, ResponsesRequest]] = None, ): if not model_type: raise ValueError("Model type must be specified") @@ -631,15 +723,15 @@ def __init__( if chat_template_kwargs.get("force_nonempty_content") is True: kwargs["force_nonempty_content"] = True - # K2-v3 selects its token pair via reasoning_effort. The OpenAI server - # pops reasoning_effort out of chat_template_kwargs and promotes it to - # request.reasoning_effort (see serving_chat.py); fall back to the - # kwargs dict for callers that bypass that normalization. + # K2-v3 selects its token pair via reasoning_effort. Chat Completions + # exposes it at the top level, while Responses nests it under reasoning. + # Fall back to chat_template_kwargs for callers that bypass normalization. if model_type.lower() in ("k2_v3", "k2_v3_legacy"): - effort = ( - getattr(request, "reasoning_effort", None) - or chat_template_kwargs.get("reasoning_effort") - ) + effort = getattr(request, "reasoning_effort", None) + if not effort: + reasoning = getattr(request, "reasoning", None) + effort = getattr(reasoning, "effort", None) + effort = effort or chat_template_kwargs.get("reasoning_effort") if effort: kwargs["reasoning_effort"] = effort @@ -656,3 +748,12 @@ def parse_stream_chunk( """Streaming call: incremental parsing""" ret = self.detector.parse_streaming_increment(chunk_text) return ret.reasoning_text, ret.normal_text + + def parse_stream_chunk_result(self, chunk_text: str) -> StreamingParseResult: + """Streaming call preserving whether empty fields were explicitly emitted.""" + return self.detector.parse_streaming_increment(chunk_text) + + def finalize_reasoning_streaming( + self, + ) -> Optional[ReasoningParserStreamingFinalization]: + return self.detector.finalize_reasoning_streaming() diff --git a/test/registered/function_call/test_k2v3_tool_parser.py b/test/registered/function_call/test_k2v3_tool_parser.py index f1d61a66ca79..e6c46cebccdc 100644 --- a/test/registered/function_call/test_k2v3_tool_parser.py +++ b/test/registered/function_call/test_k2v3_tool_parser.py @@ -1,7 +1,6 @@ -"""Unit tests for K2V3Detector (BBQ 0518 IFM tool-call format). +"""K2-V3 and multi-format tool parser regression tests. -Ported from vLLM's K2V3ToolParser tests -(tests/entrypoints/openai/tool_parsers/test_multi_format_tool_parser.py). +Ported from LLM360/vllm PR #12. """ import json @@ -13,11 +12,12 @@ MultiFormatDetector, ) from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase register_cpu_ci(5, "stage-a-test-cpu") -def _make_tool(name, parameters=None): +def _make_tool(name: str, parameters: dict | None = None) -> Tool: return Tool( type="function", function=Function( @@ -32,326 +32,209 @@ def _make_tool(name, parameters=None): ) -class TestK2V3DetectorConstruction(unittest.TestCase): - """K2V3Detector defaults to the IFM 'xml' dialect with no delegate.""" +TOOLS = [_make_tool("get_weather"), _make_tool("get_time")] +CALL_1 = ( + "get_weather" + "city" + "Tokyo" + "" +) +CALL_2 = ( + "get_time" + "city" + "Seoul" + "" +) +GROUPED_CALL_1 = f"{CALL_1}" +GROUPED_CALLS = f"{CALL_1}{CALL_2}" - def test_default_dialect_is_xml(self): - det = K2V3Detector() - self.assertEqual(det.tool_format, "xml") - self.assertIsNone(det._delegate) - def test_subclass_of_multi_format(self): - self.assertIsInstance(K2V3Detector(), MultiFormatDetector) +def _collect_stream(detector, chunks, *, finalize=False): + normal_text = "" + calls = [] + for chunk in chunks: + result = detector.parse_streaming_increment(chunk, TOOLS) + normal_text += result.normal_text or "" + calls.extend(result.calls) + if finalize: + result = detector.finalize_streaming(TOOLS) + normal_text += result.normal_text or "" + calls.extend(result.calls) + return normal_text, calls - def test_tool_call_format_kwarg_selects_dialect(self): - det = K2V3Detector(chat_template_kwargs={"tool_call_format": "xml_typed"}) - self.assertEqual(det.tool_format, "xml_typed") - def test_tool_format_kwarg_is_rejected(self): - with self.assertRaisesRegex( - ValueError, "Unsupported argument: tool_format" - ): - K2V3Detector(chat_template_kwargs={"tool_format": "json"}) +class TestConstruction(CustomTestCase): + def test_k2_defaults_to_xml(self): + detector = K2V3Detector() + self.assertEqual(detector.tool_format, "xml") + self.assertIsNone(detector._delegate) - def test_tool_calling_format_kwarg_is_rejected(self): - with self.assertRaisesRegex( - ValueError, "Unsupported argument: tool_calling_format" - ): - K2V3Detector(chat_template_kwargs={"tool_calling_format": "xml_typed"}) + def test_tool_call_format_selects_dialect(self): + detector = K2V3Detector(chat_template_kwargs={"tool_call_format": "xml_typed"}) + self.assertEqual(detector.tool_format, "xml_typed") - def test_json_dialect_via_positional(self): - det = K2V3Detector(tool_format="json") - self.assertEqual(det.tool_format, "json") + def test_legacy_format_kwargs_are_rejected(self): + for name in ("tool_format", "tool_calling_format"): + with self.subTest(name=name), self.assertRaisesRegex( + ValueError, f"Unsupported argument: {name}" + ): + K2V3Detector(chat_template_kwargs={name: "xml"}) - def test_unknown_dialect_errors(self): - with self.assertRaisesRegex(ValueError, "Unsupported tool_format"): - K2V3Detector(tool_format="not-a-dialect") +class TestMultiFormatStreaming(CustomTestCase): + def test_streaming_emits_leading_content_and_complete_ifm_call_together(self): + detector = MultiFormatDetector(tool_format="xml") -class TestK2V3XmlExtraction(unittest.TestCase): - def setUp(self): - self.tools = [ - _make_tool( - "get_weather", - { - "type": "object", - "properties": { - "city": {"type": "string"}, - "days": {"type": "integer"}, - }, - }, - ) - ] - self.det = K2V3Detector() - - def test_has_tool_call(self): - text = ( - "get_weather" - "cityTokyo" - "" - ) - self.assertTrue(self.det.has_tool_call(text)) - self.assertFalse(self.det.has_tool_call("no markers here")) + result = detector.parse_streaming_increment("\n" + CALL_1, TOOLS) - def test_single_call(self): - text = ( - "get_weather" - "cityTokyo" - "" - ) - result = self.det.detect_and_parse(text, self.tools) + self.assertEqual(result.normal_text, "\n") self.assertEqual(len(result.calls), 1) self.assertEqual(result.calls[0].name, "get_weather") - self.assertEqual(result.calls[0].tool_index, 0) self.assertEqual(json.loads(result.calls[0].parameters), {"city": "Tokyo"}) - self.assertIsNone(result.normal_text) - def test_schema_typed_value_coercion(self): - # 'days' is declared integer in the schema -> deserialized to int 3. - text = ( - "get_weather" - "days3" - "" - ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(json.loads(result.calls[0].parameters), {"days": 3}) + def test_streaming_emits_each_complete_ifm_call_once(self): + detector = MultiFormatDetector(tool_format="xml") - def test_schema_string_preserves_argument_whitespace(self): - text = ( - "get_weather" - "city" - "\nTokyo\n" - "" + _, calls = _collect_stream( + detector, + ["" + CALL_1, CALL_2 + ""], ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(json.loads(result.calls[0].parameters), {"city": "\nTokyo\n"}) - def test_schema_integer_strips_outer_whitespace_before_coercion(self): - text = ( - "get_weather" - "days\n3\n" - "" + self.assertEqual([call.name for call in calls], ["get_weather", "get_time"]) + self.assertEqual( + [json.loads(call.parameters) for call in calls], + [{"city": "Tokyo"}, {"city": "Seoul"}], ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(json.loads(result.calls[0].parameters), {"days": 3}) + self.assertEqual([call.tool_index for call in calls], [0, 1]) - def test_no_args(self): - text = "get_weather" - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(result.calls[0].name, "get_weather") - self.assertEqual(json.loads(result.calls[0].parameters), {}) + def test_streaming_records_structured_arguments_for_finish_processing(self): + detector = MultiFormatDetector(tool_format="xml") - def test_no_markers_returns_full_text(self): - result = self.det.detect_and_parse("plain text", self.tools) - self.assertEqual(result.calls, []) - self.assertEqual(result.normal_text, "plain text") + detector.parse_streaming_increment(CALL_1, TOOLS) - def test_unknown_tool_is_forwarded(self): - text = ( - "not_registered" - "cityTokyo" - "" + self.assertEqual( + detector.prev_tool_call_arr, + [{"name": "get_weather", "arguments": {"city": "Tokyo"}}], ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(len(result.calls), 1) - self.assertEqual(result.calls[0].tool_index, -1) - self.assertEqual(result.calls[0].name, "not_registered") + self.assertEqual(detector.streamed_args_for_tool, ['{"city": "Tokyo"}']) + def test_ifm_streaming_does_not_treat_generic_tool_call_as_marker(self): + detector = MultiFormatDetector(tool_format="xml") + content = "Use literally in the documentation." -class TestK2V3XmlTyped(unittest.TestCase): - """The hint forces a value to stay a string.""" + normal_text, calls = _collect_stream(detector, [content], finalize=True) - def test_arg_type_string_keeps_numeric_as_string(self): - det = K2V3Detector(tool_format="xml_typed") - # No schema type for user_id; the inline string wins. - text = ( - "study_args" - "user_id" - "string" - "12345" - "" - ) - result = det.detect_and_parse(text, [_make_tool("study_args")]) - self.assertEqual(json.loads(result.calls[0].parameters), {"user_id": "12345"}) - - def test_arg_type_any_preserves_argument_whitespace(self): - det = K2V3Detector(tool_format="xml_typed") - text = ( - "study_args" - "notes" - "any" - "\nkeep me\n" - "" - ) - result = det.detect_and_parse(text, [_make_tool("study_args")]) - self.assertEqual( - json.loads(result.calls[0].parameters), {"notes": "\nkeep me\n"} - ) + self.assertEqual(normal_text, content) + self.assertEqual(calls, []) - def test_boolean_arg_type_strips_outer_whitespace_before_coercion(self): - det = K2V3Detector(tool_format="xml_typed") - text = ( - "study_args" - "enabled" - "boolean" - "\ntrue\n" - "" - ) - result = det.detect_and_parse(text, [_make_tool("study_args")]) - self.assertEqual(json.loads(result.calls[0].parameters), {"enabled": True}) +class TestK2GroupedToolCalls(CustomTestCase): + def test_k2_v3_streaming_requires_grouped_ifm_tool_calls(self): + detector = K2V3Detector() -class TestK2V3ReasoningPrefix(unittest.TestCase): - def setUp(self): - self.tools = [_make_tool("get_weather")] - self.det = K2V3Detector() + normal_text, calls = _collect_stream(detector, [CALL_1], finalize=True) - def test_ifm_reasoning_prefix_is_stripped(self): - text = ( - "need lookup\n" - "\n" - "get_weather" - "cityTokyo" - "\n" - "" - ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(result.normal_text, "\n") - self.assertEqual(result.calls[0].name, "get_weather") - self.assertEqual(json.loads(result.calls[0].parameters), {"city": "Tokyo"}) + self.assertEqual(normal_text, CALL_1) + self.assertEqual(calls, []) - def test_tool_only_preserves_newline_after_reasoning_prefix(self): - text = ( - "\n\n" - "\n" - "get_weather" - "cityTokyo" - "\n" - "" - ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(result.normal_text, "\n") - self.assertEqual(result.calls[0].name, "get_weather") - self.assertEqual(json.loads(result.calls[0].parameters), {"city": "Tokyo"}) + def test_k2_v3_nonstreaming_requires_grouped_ifm_tool_calls(self): + result = K2V3Detector().detect_and_parse(CALL_1, TOOLS) - def test_tool_only_preserves_whitespace_prefix_before_tool_calls(self): - text = ( - "\n" - "\n" - "get_weather" - "cityTokyo" - "\n" - "" - ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(result.normal_text, "\n") - self.assertEqual(result.calls[0].name, "get_weather") - self.assertEqual(json.loads(result.calls[0].parameters), {"city": "Tokyo"}) + self.assertEqual(result.calls, []) + self.assertEqual(result.normal_text, CALL_1) - def test_all_three_effort_blocks_are_stripped(self): - for think in ("think", "think_fast", "think_faster"): - with self.subTest(think=think): - text = ( - f"reasoning" - "get_weather" - "city" - "Tokyo" - "" + def test_k2_v3_nonstreaming_requires_closed_grouped_ifm_tool_calls(self): + incomplete_group = "" + CALL_1 + + result = K2V3Detector().detect_and_parse(incomplete_group, TOOLS) + + self.assertEqual(result.calls, []) + self.assertEqual(result.normal_text, incomplete_group) + + def test_k2_v3_nonstreaming_invalid_closed_group_preserves_full_content(self): + wrapped_contents = ("", "get_weather") + for wrapped_content in wrapped_contents: + with self.subTest(wrapped_content=wrapped_content): + model_output = ( + "Content prefix. " + f"{wrapped_content}" + " Content suffix." ) - result = self.det.detect_and_parse(text, self.tools) - self.assertIsNone(result.normal_text) - self.assertEqual(result.calls[0].name, "get_weather") - - def test_stripped_ifm_reasoning_prefix_empty_content_returns_none(self): - text = ( - "need lookup" - "" - "get_weather" - "cityTokyo" - "" - "" - ) - result = self.det.detect_and_parse(text, self.tools) - self.assertIsNone(result.normal_text) - self.assertEqual(result.calls[0].name, "get_weather") - def test_legacy_think_prefix_is_not_stripped(self): - text = ( - "legacy reasoning\n" - "get_weather" - "cityTokyo" - "" + result = K2V3Detector().detect_and_parse(model_output, TOOLS) + + self.assertEqual(result.calls, []) + self.assertEqual(result.normal_text, model_output) + + def test_k2_v3_nonstreaming_parses_only_closed_grouped_ifm_tool_calls(self): + cases = ( + (GROUPED_CALL_1, ["get_weather"]), + (GROUPED_CALLS, ["get_weather", "get_time"]), ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(result.normal_text, "legacy reasoning\n") - self.assertEqual(result.calls[0].name, "get_weather") + for grouped_calls, expected_names in cases: + with self.subTest(expected_names=expected_names): + result = K2V3Detector().detect_and_parse(grouped_calls, TOOLS) + self.assertEqual(result.normal_text, "") + self.assertEqual([call.name for call in result.calls], expected_names) -class TestK2V3JsonDialect(unittest.TestCase): - def setUp(self): - self.tools = [_make_tool("get_weather")] - self.det = K2V3Detector(tool_format="json") + def test_multi_format_nonstreaming_keeps_singular_ifm_compatibility(self): + result = MultiFormatDetector(tool_format="xml").detect_and_parse(CALL_1, TOOLS) - def test_json_object_tool_call(self): - text = ( - "" - '{"name": "get_weather", "arguments": {"city": "Tokyo"}}' - "" - ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(len(result.calls), 1) - self.assertEqual(result.calls[0].name, "get_weather") - self.assertIsNone(result.normal_text) - self.assertEqual(json.loads(result.calls[0].parameters), {"city": "Tokyo"}) + self.assertEqual([call.name for call in result.calls], ["get_weather"]) - def test_json_arguments_as_string(self): - text = ( - "" - '{"name": "get_weather", "arguments": "{\\"city\\": \\"Tokyo\\"}"}' - "" - ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual(json.loads(result.calls[0].parameters), {"city": "Tokyo"}) + def test_k2_v3_streaming_parses_grouped_ifm_tool_calls(self): + normal_text, calls = _collect_stream(K2V3Detector(), [GROUPED_CALL_1]) - def test_json_list_of_tool_calls(self): - text = ( - "" - '[{"name": "get_weather", "arguments": {"city": "Tokyo"}},' - ' {"name": "get_weather", "arguments": {"city": "Osaka"}}]' - "" + self.assertEqual(normal_text, "") + self.assertEqual([call.name for call in calls], ["get_weather"]) + + def test_k2_v3_streaming_waits_for_complete_group_before_emitting_calls(self): + detector = K2V3Detector() + incomplete_group = "" + CALL_1 + + result = detector.parse_streaming_increment(incomplete_group, TOOLS) + + self.assertEqual(result.calls, []) + self.assertTrue(detector.has_pending_streaming_output()) + final_result = detector.finalize_streaming(TOOLS) + self.assertEqual(final_result.normal_text, incomplete_group) + self.assertEqual(final_result.calls, []) + self.assertFalse(detector.has_pending_streaming_output()) + + def test_k2_v3_streaming_preserves_multiple_complete_calls(self): + _, calls = _collect_stream(K2V3Detector(), [GROUPED_CALLS]) + + self.assertEqual([call.name for call in calls], ["get_weather", "get_time"]) + self.assertEqual([call.tool_index for call in calls], [0, 1]) + + def test_grouped_json_dialect_uses_schema_coercion(self): + output = ( + "" + '{"name":"get_weather","arguments":{"city":12345}}' + "" ) - result = self.det.detect_and_parse(text, self.tools) - self.assertEqual([c.name for c in result.calls], ["get_weather", "get_weather"]) - self.assertEqual(json.loads(result.calls[1].parameters), {"city": "Osaka"}) + result = K2V3Detector(tool_format="json").detect_and_parse(output, TOOLS) -def _collect_stream(detector, tools, chunks): - """Feed ``chunks`` to a detector's streaming parser and reassemble the - per-tool name + parameters, plus the concatenated normal text. Mirrors the - GLM detector streaming tests' collection helper.""" - by_index = {} - order = [] - normal_text = "" - for chunk in chunks: - result = detector.parse_streaming_increment(chunk, tools) - normal_text += result.normal_text or "" - for item in result.calls: - idx = item.tool_index - if idx not in by_index: - by_index[idx] = {"name": None, "parameters": ""} - order.append(idx) - if item.name: - by_index[idx]["name"] = item.name - if item.parameters: - by_index[idx]["parameters"] += item.parameters - return normal_text, [by_index[i] for i in order] - - -class TestK2V3XmlStreaming(unittest.TestCase): - """The IFM xml/xml_typed dialects stream incrementally (name then args).""" - - def setUp(self): - self.tools = [ + self.assertEqual(json.loads(result.calls[0].parameters), {"city": "12345"}) + + +class TestK2RegressionCoverage(CustomTestCase): + """Retain parser coverage that predates the grouped-call regressions.""" + + @staticmethod + def _group(*calls: str) -> str: + return f"{''.join(calls)}" + + def test_construction_and_unknown_dialect(self): + self.assertIsInstance(K2V3Detector(), MultiFormatDetector) + self.assertEqual(K2V3Detector(tool_format="json").tool_format, "json") + with self.assertRaisesRegex(ValueError, "Unsupported tool_format"): + K2V3Detector(tool_format="not-a-dialect") + + def test_xml_schema_coercion_no_args_and_unknown_tools(self): + tools = [ _make_tool( "get_weather", { @@ -363,399 +246,257 @@ def setUp(self): }, ) ] - self.block = ( - "get_weather" - "cityTokyo" - "" + cases = ( + ( + "get_weather" + "city" + "\nTokyo\n" + "", + {"city": "\nTokyo\n"}, + 0, + ), + ( + "get_weather" + "days" + "\n3\n" + "", + {"days": 3}, + 0, + ), + ("get_weather", {}, 0), + ( + "not_registered" + "city" + "Tokyo" + "", + {"city": "Tokyo"}, + -1, + ), ) - def _assert_single_weather(self, normal_text, calls, expected_args): - self.assertEqual(normal_text, "") - self.assertEqual(len(calls), 1) - self.assertEqual(calls[0]["name"], "get_weather") - self.assertEqual(json.loads(calls[0]["parameters"]), expected_args) - # Streaming must reassemble to the same arguments the non-stream parser - # produces. - nonstream = K2V3Detector().detect_and_parse(self.block, self.tools) - self.assertEqual( - json.loads(calls[0]["parameters"]), - json.loads(nonstream.calls[0].parameters), - ) + for call, expected_arguments, expected_index in cases: + with self.subTest(expected_arguments=expected_arguments): + result = K2V3Detector().detect_and_parse(self._group(call), tools) + self.assertEqual(len(result.calls), 1) + self.assertEqual( + json.loads(result.calls[0].parameters), expected_arguments + ) + self.assertEqual(result.calls[0].tool_index, expected_index) - def test_whole_block_one_chunk(self): - normal, calls = _collect_stream(K2V3Detector(), self.tools, [self.block]) - self._assert_single_weather(normal, calls, {"city": "Tokyo"}) - - def test_char_by_char(self): - chunks = list(self.block) # one character per chunk - normal, calls = _collect_stream(K2V3Detector(), self.tools, chunks) - self._assert_single_weather(normal, calls, {"city": "Tokyo"}) - - def test_first_param_item_is_name_only(self): - det = K2V3Detector() - # Send the name-bearing prefix, then the rest. - det.parse_streaming_increment("get_weather", self.tools) - first = det.parse_streaming_increment( - "city", self.tools + def test_xml_typed_inline_coercion(self): + tools = [_make_tool("study_args")] + cases = ( + ( + "user_id", + "string", + "12345", + {"user_id": "12345"}, + ), + ( + "notes", + "any", + "\nkeep me\n", + {"notes": "\nkeep me\n"}, + ), + ( + "enabled", + "boolean", + "\ntrue\n", + {"enabled": True}, + ), ) - # The name item is emitted with empty parameters before any args. - name_items = [c for c in first.calls if c.name] - self.assertTrue(any(c.name == "get_weather" for c in name_items)) - self.assertTrue(all(c.parameters == "" for c in name_items)) - - def test_split_at_awkward_tag_boundaries(self): - chunks = [ - "get_wea", - "thercityTo", - "kyo", - ] - normal, calls = _collect_stream(K2V3Detector(), self.tools, chunks) - self._assert_single_weather(normal, calls, {"city": "Tokyo"}) - def test_schema_typed_integer(self): - block = ( - "get_weather" - "days3" - "" - ) - normal, calls = _collect_stream(K2V3Detector(), self.tools, list(block)) - self.assertEqual(len(calls), 1) - self.assertEqual(json.loads(calls[0]["parameters"]), {"days": 3}) - - def test_no_args(self): - block = "get_weather" - normal, calls = _collect_stream(K2V3Detector(), self.tools, list(block)) - self.assertEqual(len(calls), 1) - self.assertEqual(calls[0]["name"], "get_weather") - self.assertEqual(json.loads(calls[0]["parameters"]), {}) - - def test_empty_value_object_typed_stays_valid_json(self): - # An empty on a non-string (array) type is - # malformed input; streaming must still reassemble to parseable JSON. - tools = [_make_tool("todo", {"type": "object", "properties": {"items": {"type": "array"}}})] - block = ( - "todo" - "items" - "" - ) - normal, calls = _collect_stream(K2V3Detector(), tools, list(block)) - self.assertEqual(len(calls), 1) - # Reassembles to valid JSON (does not produce '{"items": }'). - self.assertEqual(json.loads(calls[0]["parameters"]), {"items": ""}) + for key, arg_type, value, expected_arguments in cases: + with self.subTest(arg_type=arg_type): + call = ( + "study_args" + f"{key}" + f"{arg_type}" + f"{value}" + "" + ) + result = K2V3Detector(tool_format="xml_typed").detect_and_parse( + self._group(call), tools + ) + self.assertEqual( + json.loads(result.calls[0].parameters), expected_arguments + ) - def test_multiple_calls_separate_chunks(self): - chunk_a = ( - "get_weather" - "cityTokyo" - "" - ) - chunk_b = ( - "get_weather" - "cityOsaka" - "" - ) - normal, calls = _collect_stream( - K2V3Detector(), self.tools, [chunk_a, chunk_b] + def test_json_argument_shapes(self): + detector = K2V3Detector(tool_format="json") + cases = ( + ( + '{"name":"get_weather","arguments":{"city":"Tokyo"}}', + [{"city": "Tokyo"}], + ), + ( + '{"name":"get_weather","arguments":"{\\"city\\":\\"Tokyo\\"}"}', + [{"city": "Tokyo"}], + ), + ( + '[{"name":"get_weather","arguments":{"city":"Tokyo"}},' + '{"name":"get_weather","arguments":{"city":"Osaka"}}]', + [{"city": "Tokyo"}, {"city": "Osaka"}], + ), ) - self.assertEqual(len(calls), 2) - self.assertEqual(json.loads(calls[0]["parameters"]), {"city": "Tokyo"}) - self.assertEqual(json.loads(calls[1]["parameters"]), {"city": "Osaka"}) - def test_multiple_calls_same_chunk(self): - combined = ( - "get_weather" - "cityTokyo" - "" - "get_weather" - "cityOsaka" - "" - ) - normal, calls = _collect_stream(K2V3Detector(), self.tools, [combined]) - self.assertEqual(normal, "") - self.assertEqual(len(calls), 2) - self.assertEqual(json.loads(calls[0]["parameters"]), {"city": "Tokyo"}) - self.assertEqual(json.loads(calls[1]["parameters"]), {"city": "Osaka"}) - - def test_multiple_calls_char_by_char(self): - combined = ( - "get_weather" - "cityTokyo" - "" - "get_weather" - "cityOsaka" - "" - ) - normal, calls = _collect_stream(K2V3Detector(), self.tools, list(combined)) - self.assertEqual([c["name"] for c in calls], ["get_weather", "get_weather"]) - self.assertEqual(json.loads(calls[0]["parameters"]), {"city": "Tokyo"}) - self.assertEqual(json.loads(calls[1]["parameters"]), {"city": "Osaka"}) - - def test_reasoning_prefix_and_wrapper_stripped(self): - wire = ( - "need lookup\n" - "\n" - "get_weather" - "cityTokyo" - "\n" - "" - ) - normal, calls = _collect_stream(K2V3Detector(), self.tools, list(wire)) - # No reasoning text or structural tokens leak; only incidental - # boundary whitespace may pass through when fed character-by-character. - self.assertEqual(normal.strip(), "") - self.assertEqual(len(calls), 1) - self.assertEqual(calls[0]["name"], "get_weather") - self.assertEqual(json.loads(calls[0]["parameters"]), {"city": "Tokyo"}) - - def test_normal_text_before_tool_call_is_emitted(self): - wire = "Sure, let me check.get_weather" - normal, calls = _collect_stream(K2V3Detector(), self.tools, list(wire)) - self.assertEqual(normal, "Sure, let me check.") - self.assertEqual(calls[0]["name"], "get_weather") - - def test_whitespace_before_tool_call_is_emitted(self): - wire = "\nget_weather" - normal, calls = _collect_stream(K2V3Detector(), self.tools, list(wire)) - self.assertEqual(normal, "\n") - self.assertEqual(calls[0]["name"], "get_weather") - - def test_normal_text_after_tool_call_is_buffered(self): - det = K2V3Detector() - first = det.parse_streaming_increment( - f"Before {self.block}After", self.tools - ) - self.assertEqual(first.normal_text, "Before ") - self.assertTrue(any(call.name == "get_weather" for call in first.calls)) + for payload, expected_arguments in cases: + with self.subTest(expected_arguments=expected_arguments): + call = f"{payload}" + result = detector.detect_and_parse(self._group(call), TOOLS) + self.assertEqual( + [json.loads(item.parameters) for item in result.calls], + expected_arguments, + ) - second = det.parse_streaming_increment(" later", self.tools) - self.assertEqual(second.normal_text, "After later") - self.assertEqual(second.calls, []) + def test_reasoning_prefix_and_whitespace_handling(self): + for effort in ("think", "think_fast", "think_faster"): + with self.subTest(effort=effort): + output = f"need lookup" + GROUPED_CALL_1 + result = K2V3Detector().detect_and_parse(output, TOOLS) + self.assertEqual(result.normal_text, "") + + output = "need lookup\n" + GROUPED_CALL_1 + result = K2V3Detector().detect_and_parse(output, TOOLS) + self.assertEqual(result.normal_text, "") + + legacy_prefix = "legacy reasoning\n" + result = K2V3Detector().detect_and_parse(legacy_prefix + GROUPED_CALL_1, TOOLS) + self.assertEqual(result.normal_text, legacy_prefix) + + def test_grouped_streaming_handles_character_and_awkward_splits(self): + split_points = ( + list(GROUPED_CALL_1), + [ + "get_wea", + "thercityTo", + "kyo", + ], + ) + + for chunks in split_points: + with self.subTest(chunk_count=len(chunks)): + normal_text, calls = _collect_stream(K2V3Detector(), chunks) + self.assertEqual(normal_text, "") + self.assertEqual([call.name for call in calls], ["get_weather"]) + self.assertEqual(json.loads(calls[0].parameters), {"city": "Tokyo"}) - def test_normal_text_between_tool_calls_splits_results(self): - det = K2V3Detector() - block_b = ( - "get_weather" - "cityOsaka" + def test_grouped_streaming_no_args_duplicate_calls_and_malformed_value(self): + no_args = "get_weather" + duplicate = CALL_1 + CALL_1.replace("Tokyo", "Osaka") + array_tools = [ + _make_tool( + "todo", + { + "type": "object", + "properties": {"items": {"type": "array"}}, + }, + ) + ] + empty_array = ( + "todo" + "items" + "" "" ) - first = det.parse_streaming_increment( - f"{self.block}Between{block_b}", self.tools - ) - self.assertEqual(first.normal_text, "") - self.assertEqual( - [call.name for call in first.calls if call.name], ["get_weather"] - ) - - second = det.parse_streaming_increment("Tail", self.tools) - self.assertEqual(second.normal_text, "Between") - self.assertEqual( - [call.name for call in second.calls if call.name], ["get_weather"] - ) - - third = det.parse_streaming_increment(" done", self.tools) - self.assertEqual(third.normal_text, "Tail done") - self.assertEqual(third.calls, []) - -class TestK2V3XmlTypedStreaming(unittest.TestCase): - """The inline hint forces a value's type while streaming.""" + _, calls = _collect_stream(K2V3Detector(), [self._group(no_args)]) + self.assertEqual(json.loads(calls[0].parameters), {}) - def test_inline_string_keeps_numeric_as_string(self): - det = K2V3Detector(tool_format="xml_typed") - tools = [_make_tool("study_args")] - block = ( - "study_args" - "user_id" - "string" - "12345" - "" - ) - normal, calls = _collect_stream(det, tools, list(block)) - self.assertEqual(len(calls), 1) - self.assertEqual(json.loads(calls[0]["parameters"]), {"user_id": "12345"}) - # Matches the non-stream coercion. - nonstream = K2V3Detector(tool_format="xml_typed").detect_and_parse( - block, tools - ) + _, calls = _collect_stream(K2V3Detector(), [self._group(duplicate)]) + self.assertEqual([call.tool_index for call in calls], [0, 1]) self.assertEqual( - json.loads(calls[0]["parameters"]), - json.loads(nonstream.calls[0].parameters), + [json.loads(call.parameters) for call in calls], + [{"city": "Tokyo"}, {"city": "Osaka"}], ) - def test_inline_any_preserves_argument_whitespace(self): - det = K2V3Detector(tool_format="xml_typed") - tools = [_make_tool("study_args")] - block = ( - "study_args" - "notes" - "any" - "\nkeep me\n" + result = K2V3Detector().detect_and_parse(self._group(empty_array), array_tools) + self.assertEqual(json.loads(result.calls[0].parameters), {"items": ""}) + + def test_grouped_streaming_preserves_prefix_and_schema_coercion(self): + tools = [ + _make_tool( + "get_weather", + { + "type": "object", + "properties": {"days": {"type": "integer"}}, + }, + ) + ] + call = ( + "get_weather" + "days" + "3" "" ) - normal, calls = _collect_stream(det, tools, list(block)) - self.assertEqual(len(calls), 1) - self.assertEqual( - json.loads(calls[0]["parameters"]), {"notes": "\nkeep me\n"} - ) - -class TestK2V3JsonStreaming(unittest.TestCase): - """The IFM json dialect streams at tool-call-block granularity.""" + normal_text = "" + calls = [] + detector = K2V3Detector() + for chunk in "I will check. " + self._group(call): + result = detector.parse_streaming_increment(chunk, tools) + normal_text += result.normal_text or "" + calls.extend(result.calls) - def setUp(self): - self.tools = [_make_tool("get_weather")] + self.assertEqual(normal_text, "I will check. ") + self.assertEqual(json.loads(calls[0].parameters), {"days": 3}) - def test_single_object_block(self): - det = K2V3Detector(tool_format="json") - block = ( - "" - '{"name": "get_weather", "arguments": {"city": "Tokyo"}}' - "" + def test_grouped_json_streaming_waits_for_complete_wrapper(self): + payload = ( + '{"name":"get_weather",' + '"arguments":{"city":"Tokyo"}}' ) - normal, calls = _collect_stream(det, self.tools, list(block)) - self.assertEqual(len(calls), 1) - self.assertEqual(calls[0]["name"], "get_weather") - self.assertEqual(json.loads(calls[0]["parameters"]), {"city": "Tokyo"}) - - def test_emits_only_after_block_completes(self): - det = K2V3Detector(tool_format="json") - # Partial block: nothing emitted yet. - partial = det.parse_streaming_increment( - '{"name": "get_weather", "argum', self.tools - ) - self.assertEqual(partial.calls, []) - # Completing the block emits the call. - rest = det.parse_streaming_increment( - 'ents": {"city": "Tokyo"}}', self.tools - ) - names = [c.name for c in rest.calls if c.name] - self.assertIn("get_weather", names) - - def test_list_of_tool_calls(self): - det = K2V3Detector(tool_format="json") - block = ( - "" - '[{"name": "get_weather", "arguments": {"city": "Tokyo"}},' - ' {"name": "get_weather", "arguments": {"city": "Osaka"}}]' - "" - ) - normal, calls = _collect_stream(det, self.tools, list(block)) - self.assertEqual([c["name"] for c in calls], ["get_weather", "get_weather"]) - self.assertEqual(json.loads(calls[0]["parameters"]), {"city": "Tokyo"}) - self.assertEqual(json.loads(calls[1]["parameters"]), {"city": "Osaka"}) - + grouped = self._group(payload) + detector = K2V3Detector(tool_format="json") -class TestK2V3StreamingRegistry(unittest.TestCase): - """FunctionCallParser('k2_v3') streams tool calls end-to-end.""" + partial = detector.parse_streaming_increment(grouped[:-1], TOOLS) + self.assertEqual(partial.calls, []) + completed = detector.parse_streaming_increment(grouped[-1], TOOLS) + self.assertEqual([call.name for call in completed.calls], ["get_weather"]) + self.assertEqual(json.loads(completed.calls[0].parameters), {"city": "Tokyo"}) - def test_parse_stream_chunk_streams(self): + def test_function_call_parser_registry_and_streaming(self): from sglang.srt.function_call.function_call_parser import FunctionCallParser - tools = [_make_tool("get_weather")] parser = FunctionCallParser( - tools=tools, + tools=TOOLS, tool_call_parser="k2_v3", chat_template_kwargs={"tool_call_format": "xml"}, ) - block = ( - "get_weather" - "cityTokyo" - "" - ) - name = None - params = "" - for ch in list(block): - _, calls = parser.parse_stream_chunk(ch) - for c in calls: - if c.name: - name = c.name - if c.parameters: - params += c.parameters - self.assertEqual(name, "get_weather") - self.assertEqual(json.loads(params), {"city": "Tokyo"}) - - -class TestK2V3RegistryWiring(unittest.TestCase): - """FunctionCallParser resolves 'k2_v3' to K2V3Detector end-to-end.""" - - def test_registry_builds_k2v3_detector(self): + self.assertIsInstance(parser.detector, K2V3Detector) + + streamed_calls = [] + for chunk in GROUPED_CALL_1: + _, calls = parser.parse_stream_chunk(chunk) + streamed_calls.extend(calls) + self.assertEqual([call.name for call in streamed_calls], ["get_weather"]) + self.assertEqual(json.loads(streamed_calls[0].parameters), {"city": "Tokyo"}) + + parser = FunctionCallParser(tools=TOOLS, tool_call_parser="k2_v3") + normal_text, calls = parser.parse_non_stream(GROUPED_CALL_1) + self.assertEqual(normal_text, "") + self.assertEqual([call.name for call in calls], ["get_weather"]) + + def test_function_call_parser_dialect_and_alias_validation(self): from sglang.srt.function_call.function_call_parser import FunctionCallParser - tools = [_make_tool("get_weather")] parser = FunctionCallParser( - tools=tools, + tools=TOOLS, tool_call_parser="k2_v3", - chat_template_kwargs={"tool_call_format": "xml"}, + chat_template_kwargs={"tool_call_format": "json"}, ) - self.assertIsInstance(parser.detector, K2V3Detector) - self.assertEqual(parser.detector.tool_format, "xml") - - def test_registry_uses_tool_call_format_dialects_from_template(self): - from sglang.srt.function_call.function_call_parser import FunctionCallParser - - tools = [_make_tool("get_weather")] - cases = { - "json": ( - '{"name": "get_weather", ' - '"arguments": {"city": "Tokyo"}}' - ), - "xml_typed": ( - "get_weather" - "city" - "string" - "Tokyo" - "" - ), - } + self.assertEqual(parser.detector.tool_format, "json") - for dialect, wire_output in cases.items(): - with self.subTest(dialect=dialect): - parser = FunctionCallParser( - tools=tools, + for alias in ("tool_format", "tool_calling_format"): + with self.subTest(alias=alias), self.assertRaisesRegex( + ValueError, f"Unsupported argument: {alias}" + ): + FunctionCallParser( + tools=TOOLS, tool_call_parser="k2_v3", - chat_template_kwargs={"tool_call_format": dialect}, + chat_template_kwargs={alias: "json"}, ) - self.assertIsInstance(parser.detector, K2V3Detector) - self.assertEqual(parser.detector.tool_format, dialect) - - normal_text, calls = parser.parse_non_stream(wire_output) - self.assertIsNone(normal_text) - self.assertEqual(len(calls), 1) - self.assertEqual(calls[0].name, "get_weather") - self.assertEqual(json.loads(calls[0].parameters), {"city": "Tokyo"}) - - def test_registry_rejects_alias_format_kwargs(self): - from sglang.srt.function_call.function_call_parser import FunctionCallParser - - tools = [_make_tool("get_weather")] - for key in ("tool_format", "tool_calling_format"): - with self.subTest(key=key): - with self.assertRaisesRegex( - ValueError, f"Unsupported argument: {key}" - ): - FunctionCallParser( - tools=tools, - tool_call_parser="k2_v3", - chat_template_kwargs={key: "json"}, - ) - - def test_full_pipeline_non_stream(self): - from sglang.srt.function_call.function_call_parser import FunctionCallParser - - tools = [_make_tool("get_weather")] - parser = FunctionCallParser(tools=tools, tool_call_parser="k2_v3") - wire_output = ( - "looking it up" - "get_weather" - "cityTokyo" - "" - ) - normal_text, calls = parser.parse_non_stream(wire_output) - self.assertEqual(len(calls), 1) - self.assertEqual(calls[0].name, "get_weather") - self.assertEqual(json.loads(calls[0].parameters), {"city": "Tokyo"}) - self.assertIsNone(normal_text) if __name__ == "__main__": diff --git a/test/registered/openai_server/basic/test_serving_chat.py b/test/registered/openai_server/basic/test_serving_chat.py index e495b34f180a..5c48112e42c8 100644 --- a/test/registered/openai_server/basic/test_serving_chat.py +++ b/test/registered/openai_server/basic/test_serving_chat.py @@ -344,6 +344,500 @@ async def collect_chunks(): self.assertTrue(tool_deltas) self.assertEqual(tool_deltas[0][0]["function"]["name"], "get_weather") + def test_k2v3_nonstreaming_requires_a_valid_closed_plural_wrapper(self): + self.chat.tool_call_parser = "k2_v3" + tools = ChatCompletionRequest( + model="x", + messages=[], + tools=[ + { + "type": "function", + "function": { + "name": "get_weather", + "parameters": {"type": "object", "properties": {}}, + }, + } + ], + ).tools + singular = "get_weather" + invalid_group = ( + "Content get_weather" + " suffix" + ) + for output in (singular, invalid_group): + with self.subTest(output=output): + result = self.chat._process_tool_calls( + output, + tools, + {"type": "stop", "matched": None}, + tool_choice="auto", + chat_template_kwargs={"tool_call_format": "xml"}, + ) + self.assertFalse(result.tool_calls) + self.assertEqual(result.remaining_text, output) + self.assertEqual(result.finish_reason["type"], "stop") + + grouped = f"{singular}" + result = self.chat._process_tool_calls( + grouped, + tools, + {"type": "stop", "matched": None}, + tool_choice="auto", + chat_template_kwargs={"tool_call_format": "xml"}, + ) + self.assertEqual(result.remaining_text, "") + self.assertEqual(result.finish_reason["type"], "tool_calls") + self.assertEqual(result.tool_calls[0].function.name, "get_weather") + + result = self.chat._process_tool_calls( + grouped, + tools, + {"type": "length", "matched": None}, + tool_choice="auto", + chat_template_kwargs={"tool_call_format": "xml"}, + ) + self.assertEqual(result.remaining_text, "") + self.assertEqual(result.finish_reason["type"], "tool_calls") + self.assertEqual(result.tool_calls[0].function.name, "get_weather") + + def test_k2v3_streaming_reasoning_and_tool_boundary_matrix(self): + """Port the serving-path matrix from LLM360/vllm PR #12.""" + tool_call = ( + "" + "get_weather" + "city" + "Tokyo" + "" + "" + ) + singular_tool_call = tool_call.removeprefix("").removesuffix( + "" + ) + cases = [ + ( + "same_delta", + [(f"\n{tool_call}", "stop")], + "\n", + [""], + "tool_calls", + ), + ( + "split_deltas", + [("", None), (f"\n{tool_call}", "stop")], + "\n", + [""], + "tool_calls", + ), + ( + "missing_close_stop", + [(tool_call, "stop")], + "", + [""], + "tool_calls", + ), + ( + "missing_close_length", + [(tool_call, "length")], + "", + [""], + "tool_calls", + ), + ( + "missing_close_reasoning_and_tool", + [(f"Need lookup. {tool_call}", "stop")], + "", + ["Need lookup. "], + "tool_calls", + ), + ( + "plain_missing_close_stop", + [("Plain answer ", None), ("without close.", "stop")], + "Plain answer without close.", + [""], + "stop", + ), + ( + "plain_missing_close_length", + [("Plain answer ", None), ("without close.", "length")], + "Plain answer without close.", + [""], + "length", + ), + ( + "missing_close_incomplete_length", + [ + ( + "Need lookup. " "get_weather", + "length", + ) + ], + "get_weather", + ["Need lookup. "], + "length", + ), + ( + "explicit_close_incomplete_tool_is_content", + [ + ( + "Need lookup." + "get_weather", + "stop", + ) + ], + "get_weather", + ["Need lookup."], + "stop", + ), + ( + "delayed_explicit_close", + [ + (f"Maybe {tool_call}", None), + (f" reconsider{tool_call}", "stop"), + ], + "", + [f"Maybe {tool_call} reconsider"], + "tool_calls", + ), + ( + "multiple_close_tokens", + [("Need lookup.Answer", "stop")], + "Answer", + ["Need lookup."], + "stop", + ), + ( + "singular_tool_tag_remains_content", + [(f"Need lookup. {singular_tool_call}", "stop")], + f"Need lookup. {singular_tool_call}", + [""], + "stop", + ), + ] + + for ( + name, + model_deltas, + expected_content, + expected_reasoning, + expected_finish, + ) in cases: + with self.subTest(name=name): + self.chat.reasoning_parser = "k2_v3" + self.chat.tool_call_parser = "k2_v3" + self.template_manager.force_reasoning = False + request = ChatCompletionRequest( + model="x", + messages=[{"role": "user", "content": "Weather?"}], + tools=[ + { + "type": "function", + "function": { + "name": "get_weather", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + }, + } + ], + tool_choice="auto", + stream=True, + chat_template_kwargs={"tool_call_format": "xml"}, + ) + + async def generate(): + cumulative_text = "" + for delta_text, finish_type in model_deltas: + cumulative_text += delta_text + yield { + "text": cumulative_text, + "meta_info": { + "id": "chatcmpl-test", + "prompt_tokens": 1, + "completion_tokens": len(cumulative_text), + "reasoning_tokens": 0, + "cached_tokens": 0, + "finish_reason": ( + {"type": finish_type, "matched": None} + if finish_type + else None + ), + }, + "index": 0, + } + + self.tm.generate_request.return_value = generate() + + async def collect_chunks(): + return [ + chunk + async for chunk in self.chat._generate_chat_stream( + GenerateReqInput(input_ids=[1], stream=True), + request, + self.fastapi_request, + ) + ] + + chunks = get_or_create_event_loop().run_until_complete(collect_chunks()) + payloads = [ + json.loads(chunk[len("data: ") :]) + for chunk in chunks + if chunk.startswith("data: {") + ] + choices = [ + payload["choices"][0] for payload in payloads if payload["choices"] + ] + reasoning = [ + choice["delta"]["reasoning_content"] + for choice in choices + if choice["delta"].get("reasoning_content") is not None + ] + content = "".join( + choice["delta"].get("content") or "" for choice in choices + ) + finish = next( + choice["finish_reason"] + for choice in choices + if choice["finish_reason"] is not None + ) + tool_calls = [ + tool + for choice in choices + for tool in choice["delta"].get("tool_calls") or [] + ] + + self.assertEqual(reasoning, expected_reasoning) + self.assertEqual(content, expected_content) + self.assertEqual(finish, expected_finish) + if expected_finish == "tool_calls": + self.assertEqual(len(tool_calls), 1) + self.assertEqual(tool_calls[0]["function"]["name"], "get_weather") + self.assertEqual( + json.loads(tool_calls[0]["function"]["arguments"]), + {"city": "Tokyo"}, + ) + else: + self.assertEqual(tool_calls, []) + + def test_k2v3_streaming_abort_does_not_finalize_held_output(self): + self.chat.reasoning_parser = "k2_v3" + self.chat.tool_call_parser = "k2_v3" + self.template_manager.force_reasoning = False + request = ChatCompletionRequest( + model="x", + messages=[{"role": "user", "content": "Weather?"}], + tools=[ + { + "type": "function", + "function": { + "name": "get_weather", + "parameters": {"type": "object", "properties": {}}, + }, + } + ], + stream=True, + chat_template_kwargs={"tool_call_format": "xml"}, + ) + + async def generate(): + yield { + "text": "get_weather", + "meta_info": { + "id": "chatcmpl-test", + "prompt_tokens": 1, + "completion_tokens": 1, + "reasoning_tokens": 0, + "cached_tokens": 0, + "finish_reason": { + "type": "abort", + "status_code": HTTPStatus.INTERNAL_SERVER_ERROR, + "message": "aborted", + }, + }, + "index": 0, + } + + self.tm.generate_request.return_value = generate() + + async def collect_chunks(): + return [ + chunk + async for chunk in self.chat._generate_chat_stream( + GenerateReqInput(input_ids=[1], stream=True), + request, + self.fastapi_request, + ) + ] + + chunks = get_or_create_event_loop().run_until_complete(collect_chunks()) + payloads = [ + json.loads(chunk[len("data: ") :]) + for chunk in chunks + if chunk.startswith("data: {") + ] + self.assertFalse( + any( + choice["delta"].get("reasoning_content") + or choice["delta"].get("content") + or choice["delta"].get("tool_calls") + for payload in payloads + for choice in payload.get("choices", []) + ) + ) + + def test_k2v3_forced_tool_streaming_drains_terminal_json(self): + """Required and named tools must emit complete arguments at the boundary.""" + forced_output_cases = [ + ( + "required_same_delta", + "required", + [ + ( + "", + '[{"name":"get_weather","parameters":{"city":"Tokyo"}}]', + "stop", + ) + ], + {"city": "Tokyo"}, + ), + ( + "required_split_deltas", + "required", + [ + ("", "", None), + ( + "", + '[{"name":"get_weather","parameters":{"city":"Tokyo"}}]', + "stop", + ), + ], + {"city": "Tokyo"}, + ), + ( + "named_same_delta", + {"type": "function", "function": {"name": "get_weather"}}, + [ + ( + "", + '[{"name":"get_weather","parameters":{"city":"Tokyo"}}]', + "stop", + ) + ], + {"city": "Tokyo"}, + ), + ( + "named_split_deltas", + {"type": "function", "function": {"name": "get_weather"}}, + [ + ("", "", None), + ( + "", + '[{"name":"get_weather","parameters":{"city":"Tokyo"}}]', + "stop", + ), + ], + {"city": "Tokyo"}, + ), + ( + "required_empty_arguments", + "required", + [("", '[{"name":"get_weather","parameters":{}}]', "stop")], + {}, + ), + ] + + for name, tool_choice, model_deltas, expected_arguments in forced_output_cases: + with self.subTest(name=name): + self.chat.reasoning_parser = "k2_v3" + self.chat.tool_call_parser = "k2_v3" + self.template_manager.force_reasoning = False + request = ChatCompletionRequest( + model="x", + messages=[{"role": "user", "content": "Weather?"}], + tools=[ + { + "type": "function", + "function": { + "name": "get_weather", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + }, + } + ], + tool_choice=tool_choice, + stream=True, + ) + + async def generate(): + cumulative_text = "" + for reasoning_delta, tool_delta, finish_type in model_deltas: + cumulative_text += reasoning_delta + tool_delta + yield { + "text": cumulative_text, + "meta_info": { + "id": "chatcmpl-test", + "prompt_tokens": 1, + "completion_tokens": len(cumulative_text), + "reasoning_tokens": 0, + "cached_tokens": 0, + "finish_reason": ( + {"type": finish_type, "matched": None} + if finish_type + else None + ), + }, + "index": 0, + } + + self.tm.generate_request.return_value = generate() + + async def collect_chunks(): + return [ + chunk + async for chunk in self.chat._generate_chat_stream( + GenerateReqInput(input_ids=[1], stream=True), + request, + self.fastapi_request, + ) + ] + + chunks = get_or_create_event_loop().run_until_complete(collect_chunks()) + payloads = [ + json.loads(chunk[len("data: ") :]) + for chunk in chunks + if chunk.startswith("data: {") + ] + choices = [ + payload["choices"][0] for payload in payloads if payload["choices"] + ] + + streamed_calls = {} + for choice in choices: + for tool_call in choice["delta"].get("tool_calls") or []: + call = streamed_calls.setdefault( + tool_call["index"], {"name": None, "arguments": ""} + ) + function = tool_call["function"] + if function.get("name") is not None: + call["name"] = function["name"] + call["arguments"] += function.get("arguments") or "" + + self.assertEqual(list(streamed_calls), [0]) + self.assertEqual(streamed_calls[0]["name"], "get_weather") + self.assertEqual( + json.loads(streamed_calls[0]["arguments"]), expected_arguments + ) + self.assertEqual( + next( + choice["finish_reason"] + for choice in choices + if choice["finish_reason"] is not None + ), + "tool_calls", + ) + def test_stop_str_isolation_between_requests(self): """Test that stop strings from one request don't affect subsequent requests. diff --git a/test/registered/unit/entrypoints/openai/test_serving_responses.py b/test/registered/unit/entrypoints/openai/test_serving_responses.py new file mode 100644 index 000000000000..86bf19798387 --- /dev/null +++ b/test/registered/unit/entrypoints/openai/test_serving_responses.py @@ -0,0 +1,281 @@ +"""Unit tests for non-Harmony Responses streaming.""" + +import unittest +from unittest.mock import AsyncMock, Mock + +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase, maybe_stub_sgl_kernel + +maybe_stub_sgl_kernel() + +from sglang.srt.entrypoints.context import SimpleContext +from sglang.srt.entrypoints.openai.protocol import ( + RequestResponseMetadata, + ResponsesRequest, + ResponsesResponse, + UsageInfo, +) +from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses +from sglang.srt.utils import get_or_create_event_loop + +register_cpu_ci(est_time=5, suite="stage-a-test-cpu") + + +class TestSimpleResponsesStreaming(CustomTestCase): + def _collect_events(self, model_deltas, *, reasoning=None): + serving = object.__new__(OpenAIServingResponses) + serving.reasoning_parser = "k2_v3" + request = ResponsesRequest( + input="test", + stream=True, + store=False, + reasoning=reasoning, + ) + context = SimpleContext() + + async def result_generator(): + cumulative_text = "" + for delta_text, finish_type in model_deltas: + cumulative_text += delta_text + context.append_output( + { + "text": cumulative_text, + "meta_info": { + "finish_reason": ( + {"type": finish_type} if finish_type else None + ) + }, + } + ) + yield context + + async def collect(): + return [ + event + async for event in serving._process_simple_streaming_events( + request=request, + result_generator=result_generator(), + ) + ] + + return get_or_create_event_loop().run_until_complete(collect()) + + def test_simple_streaming_skips_empty_reasoning_boundary(self): + events = self._collect_events( + [("", None), ("The answer is 42.", "stop")] + ) + + self.assertFalse(any("reasoning" in event.type for event in events)) + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.output_text.delta" + ], + ["The answer is 42."], + ) + + def test_simple_streaming_treats_combined_empty_reasoning_as_content(self): + events = self._collect_events([("The answer is 42.", "stop")]) + + self.assertFalse(any("reasoning" in event.type for event in events)) + text_delta = next( + event for event in events if event.type == "response.output_text.delta" + ) + self.assertEqual(text_delta.delta, "The answer is 42.") + + def test_simple_streaming_preserves_nonempty_reasoning_before_boundary(self): + events = self._collect_events( + [("Need to calculate.The answer is 42.", "stop")] + ) + + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.reasoning_text.delta" + ], + ["Need to calculate."], + ) + reasoning_done = next( + event for event in events if event.type == "response.reasoning_text.done" + ) + self.assertEqual(reasoning_done.text, "Need to calculate.") + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.output_text.delta" + ], + ["The answer is 42."], + ) + + def test_simple_streaming_terminal_finalization_preserves_held_content(self): + wrapper = "held tool markup" + events = self._collect_events([(f"Need lookup.{wrapper}", "stop")]) + + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.reasoning_text.delta" + ], + ["Need lookup."], + ) + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.output_text.delta" + ], + [wrapper], + ) + + def test_simple_streaming_plain_unclosed_output_finalizes_as_content(self): + answer = "Plain answer without a reasoning close." + events = self._collect_events([(answer, "stop")]) + + self.assertFalse(any("reasoning" in event.type for event in events)) + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.output_text.delta" + ], + [answer], + ) + + def test_simple_streaming_combined_delta_keeps_reasoning_and_content(self): + events = self._collect_events( + [("Held reasoningFinal answer", "stop")] + ) + + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.reasoning_text.delta" + ], + ["Held reasoning"], + ) + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.output_text.delta" + ], + ["Final answer"], + ) + + def test_simple_streaming_uses_responses_reasoning_effort(self): + events = self._collect_events( + [("Fast reasoningFinal answer", "stop")], + reasoning={"effort": "medium"}, + ) + + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.reasoning_text.delta" + ], + ["Fast reasoning"], + ) + self.assertEqual( + [ + event.delta + for event in events + if event.type == "response.output_text.delta" + ], + ["Final answer"], + ) + + def test_make_request_forwards_responses_reasoning_effort(self): + serving = object.__new__(OpenAIServingResponses) + serving.tokenizer_manager = Mock() + serving.tokenizer_manager.model_config.is_multimodal = False + serving._process_messages = Mock( + return_value=Mock(prompt_ids=[1, 2], prompt=None) + ) + request = ResponsesRequest( + model="model", + input="test", + stream=True, + store=False, + reasoning={"effort": "low"}, + ) + + get_or_create_event_loop().run_until_complete( + serving._make_request(request, prev_response=None, tokenizer=Mock()) + ) + + chat_request = serving._process_messages.call_args.args[0] + self.assertEqual(chat_request.reasoning_effort, "low") + + def test_simple_streaming_keeps_content_index_stable_across_deltas(self): + events = self._collect_events( + [ + ("First", None), + (" second", "stop"), + ] + ) + + text_deltas = [ + event for event in events if event.type == "response.output_text.delta" + ] + self.assertEqual([event.delta for event in text_deltas], ["First", " second"]) + self.assertTrue( + all( + event.content_index == 0 + for event in events + if hasattr(event, "content_index") + ) + ) + + def test_simple_streaming_abort_does_not_finalize_held_content(self): + events = self._collect_events( + [("held", "abort")] + ) + + self.assertEqual(events, []) + + def test_completed_event_uses_shared_finalization_and_maps_usage(self): + serving = object.__new__(OpenAIServingResponses) + request = ResponsesRequest(input="test", stream=True, store=False) + response = ResponsesResponse.from_request( + request=request, + sampling_params={}, + model_name="model", + created_time=123, + output=[], + status="completed", + usage=UsageInfo( + prompt_tokens=3, + completion_tokens=5, + reasoning_tokens=2, + total_tokens=8, + ), + ) + serving.responses_full_generator = AsyncMock(return_value=response) + + event = get_or_create_event_loop().run_until_complete( + serving._create_streaming_completed_event( + request=request, + sampling_params={}, + context=SimpleContext(), + model_name="model", + tokenizer=None, + request_metadata=RequestResponseMetadata(request_id="request"), + created_time=123, + ) + ) + + self.assertEqual(event.type, "response.completed") + self.assertEqual(event.response.usage.input_tokens, 3) + self.assertEqual(event.response.usage.output_tokens, 5) + self.assertEqual(event.response.usage.total_tokens, 8) + serving.responses_full_generator.assert_awaited_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/unit/parser/test_k2v3_reasoning_parser.py b/test/registered/unit/parser/test_k2v3_reasoning_parser.py index d8f3b0f7be3c..1959b81cacaa 100644 --- a/test/registered/unit/parser/test_k2v3_reasoning_parser.py +++ b/test/registered/unit/parser/test_k2v3_reasoning_parser.py @@ -1,371 +1,326 @@ -"""Unit tests for K2-v3 reasoning detectors (canonical IFM + legacy).""" +"""K2-V3 reasoning-boundary regression tests from LLM360/vllm PR #12.""" import unittest from sglang.srt.parser.reasoning_parser import ( K2V3Detector, K2V3DetectorLegacy, - ReasoningParser, ) from sglang.test.ci.ci_register import register_cpu_ci from sglang.test.test_utils import CustomTestCase register_cpu_ci(est_time=5, suite="stage-a-test-cpu") +EFFORT_TOKENS = { + "high": ("", ""), + "medium": ("", ""), + "low": ("", ""), +} +EFFORTS = tuple(EFFORT_TOKENS) -class TestK2V3DetectorTokenSelection(CustomTestCase): - """K2-v3 selects its IFM token pair based on reasoning_effort.""" - - def test_high_effort_uses_think_tokens(self): - detector = K2V3Detector(reasoning_effort="high") - self.assertEqual(detector.think_start_token, "") - self.assertEqual(detector.think_end_token, "") - - def test_medium_effort_uses_think_fast_tokens(self): - detector = K2V3Detector(reasoning_effort="medium") - self.assertEqual(detector.think_start_token, "") - self.assertEqual(detector.think_end_token, "") - - def test_low_effort_uses_think_faster_tokens(self): - detector = K2V3Detector(reasoning_effort="low") - self.assertEqual(detector.think_start_token, "") - self.assertEqual(detector.think_end_token, "") - - def test_none_effort_maps_to_high(self): - detector = K2V3Detector(reasoning_effort="none") - self.assertEqual(detector.think_start_token, "") - - def test_unknown_effort_maps_to_high(self): - detector = K2V3Detector(reasoning_effort="not-a-thing") - self.assertEqual(detector.think_start_token, "") - - def test_default_effort_is_high(self): - detector = K2V3Detector() - self.assertEqual(detector.think_start_token, "") - - def test_tool_start_token_is_ifm(self): - self.assertEqual(K2V3Detector().tool_start_token, "") - - def test_force_reasoning_false_is_rejected(self): - with self.assertRaisesRegex(ValueError, "requires force_reasoning=True"): - K2V3Detector(force_reasoning=False) +def _tool_call(name: str) -> str: + return ( + f"{name}" + "city" + "Tokyo" + "" + ) -class TestK2V3DetectorLegacyTokenSelection(CustomTestCase): - """The legacy detector selects the pre-IFM token pair based on effort.""" - def test_high_effort_uses_legacy_think_tokens(self): - detector = K2V3DetectorLegacy(reasoning_effort="high") - self.assertEqual(detector.think_start_token, "") - self.assertEqual(detector.think_end_token, "") +TOOL_CALL = _tool_call("get_weather") +SECOND_TOOL_CALL = _tool_call("get_time") +GROUPED_TOOL_CALL = f"{TOOL_CALL}" +GROUPED_TOOL_CALLS = f"{TOOL_CALL}{SECOND_TOOL_CALL}" - def test_medium_effort_uses_legacy_think_fast_tokens(self): - detector = K2V3DetectorLegacy(reasoning_effort="medium") - self.assertEqual(detector.think_start_token, "") - self.assertEqual(detector.think_end_token, "") - def test_low_effort_uses_legacy_think_faster_tokens(self): - detector = K2V3DetectorLegacy(reasoning_effort="low") - self.assertEqual(detector.think_start_token, "") - self.assertEqual(detector.think_end_token, "") +def _stream(detector: K2V3Detector, deltas: list[str]): + return [detector.parse_streaming_increment(delta) for delta in deltas] - def test_none_effort_maps_to_high(self): - detector = K2V3DetectorLegacy(reasoning_effort="none") - self.assertEqual(detector.think_start_token, "") - def test_default_effort_is_high(self): - detector = K2V3DetectorLegacy() - self.assertEqual(detector.think_start_token, "") +class TestK2V3DetectorTokenSelection(CustomTestCase): + def test_effort_tokens(self): + for effort, expected_tokens in EFFORT_TOKENS.items(): + with self.subTest(effort=effort): + detector = K2V3Detector(reasoning_effort=effort) + self.assertEqual( + (detector.think_start_token, detector.think_end_token), + expected_tokens, + ) - def test_tool_start_token_is_legacy(self): - self.assertEqual(K2V3DetectorLegacy().tool_start_token, "") + def test_default_none_and_unknown_effort_use_high(self): + for effort in (None, "none", "ultra"): + with self.subTest(effort=effort): + kwargs = {} if effort is None else {"reasoning_effort": effort} + detector = K2V3Detector(**kwargs) + self.assertEqual(detector.think_start_token, "") + self.assertEqual(detector.think_end_token, "") def test_force_reasoning_false_is_rejected(self): with self.assertRaisesRegex(ValueError, "requires force_reasoning=True"): - K2V3DetectorLegacy(force_reasoning=False) - + K2V3Detector(force_reasoning=False) -class TestK2V3DetectorToolCallSplit(CustomTestCase): - """K2-v3 exits reasoning mode when appears without an end token.""" - def test_tool_call_split_detect_and_parse(self): - for effort in ["high", "medium", "low"]: +class TestK2V3NonStreaming(CustomTestCase): + def test_nonstreaming_without_boundary_returns_content(self): + for effort in EFFORTS: + for output in ("", "The answer is 42."): + with self.subTest(effort=effort, output=output): + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + output + ) + self.assertEqual(result.reasoning_text, "") + self.assertEqual(result.normal_text, output) + + def test_nonstreaming_tool_calls_wrapper_implicitly_ends_unclosed_reasoning(self): + for effort in EFFORTS: + for reasoning in ("", "Need lookup. "): + for tool_section in (GROUPED_TOOL_CALL, GROUPED_TOOL_CALLS): + with self.subTest( + effort=effort, + reasoning=reasoning, + tool_section=tool_section, + ): + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + reasoning + tool_section + ) + self.assertEqual(result.reasoning_text, reasoning) + self.assertEqual(result.normal_text, tool_section) + + def test_nonstreaming_generated_start_with_unclosed_tool_call(self): + for effort, (start, _) in EFFORT_TOKENS.items(): with self.subTest(effort=effort): - detector = K2V3Detector(reasoning_effort=effort) - text = ( - "I'll check the file.\n" - "readfilePath" - "/tmp/f" + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + f"{start}Need lookup. {GROUPED_TOOL_CALL}" ) - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "I'll check the file.\n") - self.assertTrue(result.normal_text.startswith("")) - - def test_tool_call_split_streaming(self): - for effort in ["high", "medium", "low"]: + self.assertEqual(result.reasoning_text, "Need lookup. ") + self.assertEqual(result.normal_text, GROUPED_TOOL_CALL) + + def test_nonstreaming_incomplete_tool_calls_wrapper_is_implicit_boundary(self): + incomplete_tools = ( + "", + "get_weather", + f"{TOOL_CALL}", + ) + for effort in EFFORTS: + for incomplete_tool in incomplete_tools: + with self.subTest(effort=effort, incomplete_tool=incomplete_tool): + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + f"Need lookup. {incomplete_tool}" + ) + self.assertEqual(result.reasoning_text, "Need lookup. ") + self.assertEqual(result.normal_text, incomplete_tool) + + def test_nonstreaming_singular_tool_tag_is_not_an_implicit_boundary(self): + output = f"Need lookup. {TOOL_CALL}" + for effort in EFFORTS: with self.subTest(effort=effort): - detector = K2V3Detector(reasoning_effort=effort) - r1 = detector.parse_streaming_increment("I'll check the file.\n") - self.assertEqual(r1.reasoning_text, "I'll check the file.\n") - self.assertEqual(r1.normal_text, "") - r2 = detector.parse_streaming_increment( - "read" + result = K2V3Detector(reasoning_effort=effort).detect_and_parse(output) + self.assertEqual(result.reasoning_text, "") + self.assertEqual(result.normal_text, output) + + def test_nonstreaming_explicit_close_requires_complete_plural_tool_wrapper(self): + incomplete_plural = f"{TOOL_CALL}" + for effort, (_, end) in EFFORT_TOKENS.items(): + for tool_markup in (TOOL_CALL, incomplete_plural): + with self.subTest(effort=effort, tool_markup=tool_markup): + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + f"Need lookup.{end}{tool_markup}" + ) + self.assertEqual(result.reasoning_text, "Need lookup.") + self.assertEqual(result.normal_text, tool_markup) + + def test_nonstreaming_explicit_close_response_matrix(self): + cases = ( + ("", "", "", ""), + ("Need lookup.", "", "Need lookup.", ""), + ("", "The answer is 42.", "", "The answer is 42."), + ("Need lookup.", GROUPED_TOOL_CALL, "Need lookup.", GROUPED_TOOL_CALL), + ( + "Need lookup.", + f"Calling the tool.\n{GROUPED_TOOL_CALL}", + "Need lookup.", + f"Calling the tool.\n{GROUPED_TOOL_CALL}", + ), + ) + for effort, (_, end) in EFFORT_TOKENS.items(): + for reasoning, tail, expected_reasoning, expected_content in cases: + with self.subTest(effort=effort, reasoning=reasoning, tail=tail): + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + reasoning + end + tail + ) + self.assertEqual(result.reasoning_text, expected_reasoning) + self.assertEqual(result.normal_text, expected_content) + + def test_nonstreaming_explicit_close_takes_precedence_over_tool_marker(self): + reasoning_tool = _tool_call("consider_weather") + expected_reasoning = f"Maybe call this tool: {reasoning_tool}" + for effort, (_, end) in EFFORT_TOKENS.items(): + with self.subTest(effort=effort): + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + f"{expected_reasoning}{end}{GROUPED_TOOL_CALL}" ) - self.assertEqual(r2.normal_text, "read") - self.assertEqual(r2.reasoning_text, "") + self.assertEqual(result.reasoning_text, expected_reasoning) + self.assertEqual(result.normal_text, GROUPED_TOOL_CALL) + def test_nonstreaming_multiple_close_tokens_preserve_extra_close_as_content(self): + for effort, (_, end) in EFFORT_TOKENS.items(): + with self.subTest(effort=effort): + result = K2V3Detector(reasoning_effort=effort).detect_and_parse( + f"Need lookup.{end}{end}The answer is 42." + ) + self.assertEqual(result.reasoning_text, "Need lookup.") + self.assertEqual(result.normal_text, f"{end}The answer is 42.") -class TestK2V3DetectorLegacyToolCallSplit(CustomTestCase): - """The legacy detector exits reasoning on the legacy boundary.""" - def test_tool_call_split_detect_and_parse(self): - for effort in ["high", "medium", "low"]: +class TestK2V3Streaming(CustomTestCase): + def test_streaming_standalone_end_token_emits_empty_reasoning(self): + for effort, (_, end) in EFFORT_TOKENS.items(): with self.subTest(effort=effort): - detector = K2V3DetectorLegacy(reasoning_effort=effort) - text = ( - "I'll check the file.\n" - '\n{"name": "read"}\n' - ) - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "I'll check the file.\n") - self.assertTrue(result.normal_text.startswith("")) - - def test_tool_call_split_streaming(self): - detector = K2V3DetectorLegacy(reasoning_effort="high") - r1 = detector.parse_streaming_increment("I'll check the file.\n") - self.assertEqual(r1.reasoning_text, "I'll check the file.\n") - r2 = detector.parse_streaming_increment('\n{"name": "read"}\n') - self.assertEqual(r2.normal_text, '\n{"name": "read"}\n') - self.assertEqual(r2.reasoning_text, "") - - -class TestK2V3DetectorParsing(CustomTestCase): - """K2-v3 parses forced reasoning until the selected IFM end token.""" - - def test_high_effort_parses_end_only_output(self): - detector = K2V3Detector(reasoning_effort="high") - text = "reasoning herefinal answer" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "reasoning here") - self.assertEqual(result.normal_text, "final answer") - - def test_high_effort_strips_redundant_generated_think_start(self): - detector = K2V3Detector(reasoning_effort="high") - text = "reasoning herefinal answer" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "reasoning here") - self.assertEqual(result.normal_text, "final answer") - - def test_preserves_reasoning_and_post_think_newlines(self): - detector = K2V3Detector(reasoning_effort="high") - text = ( - "\n\n" - "get_weather" - "" - ) - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "\n") - self.assertTrue(result.normal_text.startswith("\n")) - - def test_medium_effort_parses_end_only_output(self): - detector = K2V3Detector(reasoning_effort="medium") - text = "reasoning herefinal answer" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "reasoning here") - self.assertEqual(result.normal_text, "final answer") - - def test_medium_effort_parses_think_fast_block(self): - detector = K2V3Detector(reasoning_effort="medium") - text = "reasoning herefinal answer" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "reasoning here") - self.assertEqual(result.normal_text, "final answer") - - def test_low_effort_parses_think_faster_block(self): - detector = K2V3Detector(reasoning_effort="low") - text = "ra" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "r") - self.assertEqual(result.normal_text, "a") - - def test_streaming_medium_effort(self): - detector = K2V3Detector(reasoning_effort="medium", force_reasoning=True) - r1 = detector.parse_streaming_increment("partial reason") - self.assertEqual(r1.reasoning_text, "partial reason") - r2 = detector.parse_streaming_increment("inganswer") - self.assertEqual(r2.reasoning_text, "ing") - self.assertEqual(r2.normal_text, "answer") - - def test_ignores_legacy_think_tokens(self): - """The canonical detector does not treat bare as a boundary.""" - detector = K2V3Detector(reasoning_effort="high") - result = detector.detect_and_parse("reasoningstill reasoning") - # No end token -> everything stays reasoning. - self.assertEqual(result.normal_text, "") - self.assertEqual(result.reasoning_text, "reasoningstill reasoning") - - -class TestK2V3DetectorLegacyParsing(CustomTestCase): - """The legacy detector parses the pre-IFM ... tokens natively.""" - - def test_high_effort_parses_legacy_think_block(self): - detector = K2V3DetectorLegacy(reasoning_effort="high") - text = "reasoning herefinal answer" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "reasoning here") - self.assertEqual(result.normal_text, "final answer") - - def test_high_effort_parses_legacy_end_only_output(self): - detector = K2V3DetectorLegacy(reasoning_effort="high") - text = "reasoning herefinal answer" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "reasoning here") - self.assertEqual(result.normal_text, "final answer") - - def test_medium_effort_parses_legacy_think_fast_block(self): - detector = K2V3DetectorLegacy(reasoning_effort="medium") - text = "reasoning herefinal answer" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "reasoning here") - self.assertEqual(result.normal_text, "final answer") - - def test_low_effort_parses_legacy_think_faster_block(self): - detector = K2V3DetectorLegacy(reasoning_effort="low") - text = "ra" - result = detector.detect_and_parse(text) - self.assertEqual(result.reasoning_text, "r") - self.assertEqual(result.normal_text, "a") - - def test_legacy_tokens_stream_natively(self): - """Unlike the old normalize-on-parse design, legacy streaming now works.""" - detector = K2V3DetectorLegacy(reasoning_effort="medium", force_reasoning=True) - r1 = detector.parse_streaming_increment("partial reason") - self.assertEqual(r1.reasoning_text, "partial reason") - r2 = detector.parse_streaming_increment("inganswer") - self.assertEqual(r2.reasoning_text, "ing") - self.assertEqual(r2.normal_text, "answer") - - def test_legacy_and_canonical_are_equivalent_modulo_tokens(self): - for effort in ["high", "medium", "low"]: + result = K2V3Detector( + reasoning_effort=effort + ).parse_streaming_increment(end) + self.assertTrue(result.has_reasoning_text) + self.assertEqual(result.reasoning_text, "") + self.assertFalse(result.has_normal_text) + + def test_streaming_end_token_routes_following_tool_call_to_content(self): + for effort, (_, end) in EFFORT_TOKENS.items(): + for reasoning in ("", "Need lookup"): + with self.subTest(effort=effort, reasoning=reasoning): + detector = K2V3Detector(reasoning_effort=effort) + first, second = _stream(detector, [reasoning + end, TOOL_CALL]) + self.assertTrue(first.has_reasoning_text) + self.assertEqual(first.reasoning_text, reasoning) + self.assertEqual(second.normal_text, TOOL_CALL) + + def test_streaming_missing_close_quarantines_plural_wrapper_until_finalization( + self, + ): + for effort in EFFORTS: + for tool_section in (GROUPED_TOOL_CALL, GROUPED_TOOL_CALLS): + with self.subTest(effort=effort, tool_section=tool_section): + detector = K2V3Detector(reasoning_effort=effort) + marker = "" + emitted = _stream( + detector, + [ + "Need lookup. ", + marker[:8], + marker[8:] + tool_section[len(marker) :], + ], + ) + self.assertTrue( + all(not result.has_normal_text for result in emitted) + ) + finalization = detector.finalize_reasoning_streaming() + self.assertIsNotNone(finalization) + result = finalization.result + self.assertEqual(result.reasoning_text, "Need lookup. ") + self.assertEqual(result.normal_text, tool_section) + + def test_streaming_failed_plural_marker_candidate_is_released_as_content(self): + for effort in EFFORTS: with self.subTest(effort=effort): - legacy = K2V3DetectorLegacy(reasoning_effort=effort) - canonical = K2V3Detector(reasoning_effort=effort) - legacy_text = "r" + legacy.think_end_token + "a" - canonical_text = "r" + canonical.think_end_token + "a" - self.assertEqual( - legacy.detect_and_parse(legacy_text).reasoning_text, - canonical.detect_and_parse(canonical_text).reasoning_text, + detector = K2V3Detector(reasoning_effort=effort) + _stream(detector, ["Need ", "not grouped"]) + result = detector.finalize_reasoning_streaming().result + self.assertEqual(result.reasoning_text, "") + self.assertEqual(result.normal_text, "Need not grouped") + + def test_streaming_explicit_close_releases_quarantined_wrapper_as_reasoning(self): + for effort, (_, end) in EFFORT_TOKENS.items(): + with self.subTest(effort=effort): + detector = K2V3Detector(reasoning_effort=effort) + results = _stream( + detector, + ["Maybe ", GROUPED_TOOL_CALL, f" reconsider{end}Answer"], ) + emitted = [result for result in results if result.has_reasoning_text] + self.assertEqual(len(emitted), 1) self.assertEqual( - legacy.detect_and_parse(legacy_text).normal_text, - canonical.detect_and_parse(canonical_text).normal_text, + emitted[0].reasoning_text, + f"Maybe {GROUPED_TOOL_CALL} reconsider", ) + self.assertEqual(emitted[0].normal_text, "Answer") + self.assertIsNone(detector.finalize_reasoning_streaming()) + def test_streaming_only_post_close_plural_wrapper_becomes_content(self): + for effort, (_, end) in EFFORT_TOKENS.items(): + with self.subTest(effort=effort): + detector = K2V3Detector(reasoning_effort=effort) + results = _stream( + detector, + [GROUPED_TOOL_CALL, end, GROUPED_TOOL_CALLS], + ) + self.assertEqual(results[1].reasoning_text, GROUPED_TOOL_CALL) + self.assertEqual(results[2].normal_text, GROUPED_TOOL_CALLS) -class TestK2V3ParserIntegration(CustomTestCase): - """ReasoningParser wires the k2_v3 / k2_v3_legacy model types to their detectors.""" - - def test_parser_routes_to_k2v3_detector(self): - parser = ReasoningParser(model_type="k2_v3") - self.assertIsInstance(parser.detector, K2V3Detector) - self.assertEqual(parser.detector.think_start_token, "") - - def test_parser_routes_to_k2v3_legacy_detector(self): - parser = ReasoningParser(model_type="k2_v3_legacy") - self.assertIsInstance(parser.detector, K2V3DetectorLegacy) - self.assertEqual(parser.detector.think_start_token, "") - self.assertEqual(parser.detector.tool_start_token, "") - - def test_k2v3_and_legacy_are_both_registered_for_cli(self): - keys = ReasoningParser.DetectorMap.keys() - self.assertIn("k2_v3", keys) - self.assertIn("k2_v3_legacy", keys) - - def test_parser_rejects_force_reasoning_false(self): - with self.assertRaisesRegex(ValueError, "requires force_reasoning=True"): - ReasoningParser(model_type="k2_v3", force_reasoning=False) - - def test_legacy_parser_rejects_force_reasoning_false(self): - with self.assertRaisesRegex(ValueError, "requires force_reasoning=True"): - ReasoningParser(model_type="k2_v3_legacy", force_reasoning=False) - - def test_parser_allows_force_reasoning_false_for_non_k2v3(self): - parser = ReasoningParser(model_type="qwen3", force_reasoning=False) - self.assertEqual(parser.detector.think_start_token, "") - - def test_parser_forwards_reasoning_effort_medium(self): - from sglang.srt.entrypoints.openai.protocol import ChatCompletionRequest - - req = ChatCompletionRequest( - model="k2-v3", - messages=[{"role": "user", "content": "hi"}], - chat_template_kwargs={"reasoning_effort": "medium"}, - ) - parser = ReasoningParser(model_type="k2_v3", request=req) - self.assertEqual(parser.detector.think_start_token, "") - self.assertEqual(parser.detector.think_end_token, "") - - def test_parser_forwards_reasoning_effort_low(self): - from sglang.srt.entrypoints.openai.protocol import ChatCompletionRequest - - req = ChatCompletionRequest( - model="k2-v3", - messages=[{"role": "user", "content": "hi"}], - chat_template_kwargs={"reasoning_effort": "low"}, + def test_streaming_incomplete_plural_wrapper_is_released_at_finalization(self): + incomplete_tools = ( + "", + "get_weather", ) - parser = ReasoningParser(model_type="k2_v3", request=req) - self.assertEqual(parser.detector.think_start_token, "") + for effort in EFFORTS: + for incomplete_tool in incomplete_tools: + with self.subTest(effort=effort, incomplete_tool=incomplete_tool): + detector = K2V3Detector(reasoning_effort=effort) + result = detector.parse_streaming_increment( + f"Need lookup. {incomplete_tool}" + ) + self.assertFalse(result.has_normal_text) + final = detector.finalize_reasoning_streaming().result + self.assertEqual(final.reasoning_text, "Need lookup. ") + self.assertEqual(final.normal_text, incomplete_tool) + + def test_streaming_without_boundary_finalizes_as_content(self): + for effort in EFFORTS: + with self.subTest(effort=effort): + detector = K2V3Detector(reasoning_effort=effort) + _stream(detector, ["Answer ", "without a close token."]) + final = detector.finalize_reasoning_streaming().result + self.assertEqual(final.reasoning_text, "") + self.assertEqual(final.normal_text, "Answer without a close token.") + + def test_streaming_optional_generated_start_is_not_emitted(self): + for effort, (start, _) in EFFORT_TOKENS.items(): + for deltas in ([start, "Reasoning"], [start + "Reasoning"]): + with self.subTest(effort=effort, deltas=deltas): + detector = K2V3Detector(reasoning_effort=effort) + results = _stream(detector, deltas) + self.assertTrue( + all(not result.has_reasoning_text for result in results) + ) + final = detector.finalize_reasoning_streaming().result + self.assertEqual(final.reasoning_text, "") + self.assertEqual(final.normal_text, "Reasoning") + + def test_streaming_multiple_close_tokens_use_first_boundary(self): + for effort, (_, end) in EFFORT_TOKENS.items(): + with self.subTest(effort=effort): + result = K2V3Detector( + reasoning_effort=effort + ).parse_streaming_increment(f"Reasoning{end}{end}Answer") + self.assertEqual(result.reasoning_text, "Reasoning") + self.assertEqual(result.normal_text, f"{end}Answer") + + def test_streaming_terminal_partial_plural_marker_is_finalized_as_content(self): + for effort in EFFORTS: + with self.subTest(effort=effort): + detector = K2V3Detector(reasoning_effort=effort) + detector.parse_streaming_increment("Need lookup. ") - self.assertEqual(parser.detector.think_end_token, "") - - def test_parser_reads_top_level_reasoning_effort(self): - """serving_chat.py pops reasoning_effort out of chat_template_kwargs - and moves it to request.reasoning_effort before reaching the parser. - The parser must read from the top-level field, not just the kwargs.""" - from sglang.srt.entrypoints.openai.protocol import ChatCompletionRequest - - req = ChatCompletionRequest( - model="k2-v3", - messages=[{"role": "user", "content": "hi"}], - reasoning_effort="medium", - ) - parser = ReasoningParser(model_type="k2_v3", request=req) - self.assertEqual(parser.detector.think_start_token, "") - self.assertEqual(parser.detector.think_end_token, "") - - def test_parser_ignores_reasoning_effort_for_non_k2v3(self): - """reasoning_effort kwarg must NOT be forwarded to Qwen3Detector. - - Qwen3Detector.__init__ does not accept reasoning_effort. If the - ReasoningParser guard is ever dropped, building this parser would - raise TypeError. Test the absence-of-leak property explicitly, not - just the resulting token value. - """ - from sglang.srt.entrypoints.openai.protocol import ChatCompletionRequest - - req = ChatCompletionRequest( - model="qwen3", - messages=[{"role": "user", "content": "hi"}], - chat_template_kwargs={"reasoning_effort": "medium"}, - ) - try: - parser = ReasoningParser(model_type="qwen3", request=req) - except TypeError as e: - self.fail( - "reasoning_effort was incorrectly forwarded to " - f"Qwen3Detector: {e}" - ) - self.assertEqual(parser.detector.think_start_token, "") +class TestK2V3Legacy(CustomTestCase): + def test_legacy_parser_keeps_eager_streaming_behavior(self): + detector = K2V3DetectorLegacy(reasoning_effort="medium") + first = detector.parse_streaming_increment("partial reasoning") + second = detector.parse_streaming_increment("answer") + self.assertEqual(first.reasoning_text, "partial reasoning") + self.assertEqual(second.normal_text, "answer") if __name__ == "__main__":