From 3633379572d226a425427b298f82951d6125fb14 Mon Sep 17 00:00:00 2001 From: guojn1 Date: Mon, 12 Oct 2026 13:57:39 +0800 Subject: [PATCH 1/2] fix(sglang): align GLM-5.3 sampling and reasoning stops Accept negative frequency penalties as zero for GLM-5.3 to match observed official API behavior. Preserve other models and nonnegative penalties. Default omitted generation limits to 8192 tokens and match string stops after reasoning separation, closing worker generation on a match. Cover projection, stop matching, stream cleanup and usage with regression tests. --- dingo/frontend/README.md | 25 ++ dingo/frontend/sglang_prepost.py | 43 ++- dingo/frontend/sglang_processor.py | 92 ++++++- .../tests/test_sglang_reasoning_stops_unit.py | 248 ++++++++++++++++++ 4 files changed, 398 insertions(+), 10 deletions(-) create mode 100644 dingo/frontend/tests/test_sglang_reasoning_stops_unit.py diff --git a/dingo/frontend/README.md b/dingo/frontend/README.md index b31e31da684a..477e21fcab1c 100644 --- a/dingo/frontend/README.md +++ b/dingo/frontend/README.md @@ -6,3 +6,28 @@ The API gateway for serving LLM inference requests with OpenAI-compatible HTTP and KServe gRPC endpoints. See [docs/components/frontend/](../../../../docs/components/frontend/) for documentation. + +## SGLang chat generation limits and stops + +The SGLang chat processor defaults to 8192 generated tokens when both +`max_completion_tokens` and `max_tokens` are omitted. An explicit +`max_completion_tokens` takes precedence over `max_tokens`. The limit includes +reasoning and answer tokens; reaching it returns `finish_reason: "length"` and +can leave the answer empty if all tokens were spent reasoning. + +For the exact model ID `glm-5.3` (case-insensitive, also repository-qualified +IDs such as `zai-org/GLM-5.3`), negative `frequency_penalty` values are accepted +but projected to `0.0` for sampling. This compatibility policy targets normal +completion, as observed with negative values on the official GLM-5.3 API; it +does not assert how that API implements the parameter. Nonnegative values and +other models retain their existing behavior. Arbitrary serving aliases do not +activate this policy. Reasoning remains enabled, and the generation limit +remains a fallback rather than a guarantee that every prompt will finish. + +With an active reasoning parser, string `stop` sequences apply after reasoning +separation, to the non-reasoning output (including tool syntax). A match stops +worker generation and returns `finish_reason: "stop"` (or `"tool_calls"` when +applicable). Stops split across chunks are supported; `include_stop_str_in_output` +controls whether the matched string is retained. EOS and `stop_token_ids` remain +worker-side conditions. Without a reasoning parser, including requests that set +`separate_reasoning: false`, string stops retain their worker-side behavior. diff --git a/dingo/frontend/sglang_prepost.py b/dingo/frontend/sglang_prepost.py index fa651df726e8..33f0db5b4699 100644 --- a/dingo/frontend/sglang_prepost.py +++ b/dingo/frontend/sglang_prepost.py @@ -1415,6 +1415,8 @@ def __init__( reasoning_parser_name: str | None = None, eos_token_ids: list[int] | None = None, stop_strings: set[str] | None = None, + include_stop_str_in_output: bool = False, + min_tokens: int = 0, guided_decoding: dict[str, Any] | None = None, response_format_guided_active: bool = False, tool_guided_active: bool = False, @@ -1458,6 +1460,10 @@ def __init__( self._eos_token_ids = set(eos_token_ids or []) self._stop_strings = {stop for stop in (stop_strings or set()) if stop} self._pending_stop_text = "" + self._include_stop_str_in_output = include_stop_str_in_output + self._min_tokens = min_tokens + self._generated_token_count = 0 + self.matched_stop: str | None = None self._all_token_ids: list[int] = [] # Logprob records built for chunks that emit no visible delta (e.g. @@ -1663,6 +1669,32 @@ def _filter_stop_string_delta(self, text: str, finish_reason: str | None) -> str return text[:-pending_len] return text + def _filter_answer_stop(self, text: str, finish_reason: str | None) -> str: + """Match strings after reasoning separation, including split chunks.""" + text = self._pending_stop_text + text + self._pending_stop_text = "" + if self._generated_token_count < self._min_tokens: + return text + matches = [ + (pos, -len(stop), stop) + for stop in self._stop_strings + if (pos := text.find(stop)) >= 0 + ] + if matches: + pos, _, self.matched_stop = min(matches) + end = ( + pos + len(self.matched_stop) + if self._include_stop_str_in_output + else pos + ) + return text[:end] + if not finish_reason: + pending_len = _trailing_stop_prefix_len(text, self._stop_strings) + if pending_len: + self._pending_stop_text = text[-pending_len:] + return text[:-pending_len] + return text + def _parse_reasoning_delta( self, delta_text: str, @@ -1840,8 +1872,11 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No Returns: OpenAI choice dict or ``None`` if nothing to emit yet. """ + if self.matched_stop is not None: + return None raw_ids = engine_response.get("token_ids") token_ids = raw_ids if isinstance(raw_ids, list) else list(raw_ids or []) + self._generated_token_count += len(token_ids) finish_reason = engine_response.get("finish_reason") finished = finish_reason is not None if finished: @@ -1871,7 +1906,8 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No if token_ids or finished else "" ) - delta_text = self._filter_stop_string_delta(delta_text, finish_reason) + if self.reasoning_parser is None: + delta_text = self._filter_stop_string_delta(delta_text, finish_reason) if self._is_kimi_k3 and delta_text: self._kimi_k3_raw_text_parts.append(delta_text) @@ -1903,6 +1939,10 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No len(token_ids), ) self._reasoning_token_count += reasoning_tokens + if self.reasoning_parser is not None: + normal_text = self._filter_answer_stop(normal_text, finish_reason) + if self.matched_stop is not None: + finish_reason = "stop" if self._is_kimi_k3: reasoning_text = _strip_kimi_k3_control_markers(reasoning_text) or None @@ -2155,6 +2195,7 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No if ( finish_reason and self._is_kimi_k3 + and self.matched_stop is None and not self._saw_normal_output and not self._tool_call_names ): diff --git a/dingo/frontend/sglang_processor.py b/dingo/frontend/sglang_processor.py index faf3e1811253..d45be5e9db68 100644 --- a/dingo/frontend/sglang_processor.py +++ b/dingo/frontend/sglang_processor.py @@ -51,6 +51,8 @@ logger = logging.getLogger(__name__) DEFAULT_SGLANG_STREAM_INTERVAL: int = 1 +# Bound reasoning and answer tokens together when the client omits both limits. +DEFAULT_SGLANG_MAX_TOKENS: int = 8192 # Internal carrier key mirroring the Rust constant in # lib/llm/src/protocols/openai/chat_completions.rs (INTERNAL_SGLEXT_KEY). @@ -121,8 +123,10 @@ def _normalize_eos_token_ids(value: Any) -> list[int]: return [] -def _request_stop_strings(request: dict[str, Any]) -> set[str]: - if request.get("include_stop_str_in_output") is True: +def _request_stop_strings( + request: dict[str, Any], *, match_visible: bool = False +) -> set[str]: + if request.get("include_stop_str_in_output") is True and not match_visible: return set() stop = request.get("stop") if isinstance(stop, str): @@ -376,7 +380,24 @@ def _build_dynamo_preproc( require_reasoning: bool = False, ) -> dict[str, Any]: """Build the Dynamo preprocessed request dict from request fields.""" - max_tokens = request.get("max_completion_tokens") or request.get("max_tokens") + max_tokens = request.get("max_completion_tokens") + if max_tokens is None: + max_tokens = request.get("max_tokens") + if max_tokens is None: + max_tokens = DEFAULT_SGLANG_MAX_TOKENS + + frequency_penalty = request.get("frequency_penalty", 0.0) + # GLM-5.3 compatibility policy: accept negative values without rewarding + # repetition. Official API probes complete normally with -2; this is our + # compatibility choice, not a claim about the provider's implementation. + # Match the exact model ID (also accepting a repository-qualified ID), + # rather than applying the policy to every model using the GLM parser. + if ( + model_name.rsplit("/", 1)[-1].lower() == "glm-5.3" + and isinstance(frequency_penalty, (int, float)) + and frequency_penalty < 0 + ): + frequency_penalty = 0.0 stop = request.get("stop") stop_token_ids = request.get("stop_token_ids", []) @@ -390,6 +411,12 @@ def _build_dynamo_preproc( elif stop is None: stop = [] + # The worker sees the unsplit reasoning/answer stream. String stops belong + # to the frontend when a reasoning parser is active; token stops and EOS + # remain backend-owned. + if reasoning_parser is not None: + stop = [] + # Handle logprobs logprobs_val = None logprobs = request.get("logprobs") @@ -415,7 +442,7 @@ def _build_dynamo_preproc( "sampling_options": { "n": request.get("n", 1), "presence_penalty": request.get("presence_penalty", 0.0), - "frequency_penalty": request.get("frequency_penalty", 0.0), + "frequency_penalty": frequency_penalty, "repetition_penalty": request.get("repetition_penalty", 1.0), "temperature": request.get("temperature", 1.0), "top_p": request.get("top_p", 1.0), @@ -596,7 +623,11 @@ async def _generator_inner( tool_call_parser_name=self.tool_call_parser_name, reasoning_parser_name=self.reasoning_parser_name, eos_token_ids=self.eos_token_ids, - stop_strings=_request_stop_strings(request), + stop_strings=_request_stop_strings( + request, match_visible=pre.reasoning_parser is not None + ), + include_stop_str_in_output=request.get("include_stop_str_in_output", False), + min_tokens=request.get("min_tokens", 0), guided_decoding=pre.guided_decoding, response_format_guided_active=pre.response_format_guided_active, tool_guided_active=pre.tool_guided_active, @@ -670,7 +701,11 @@ async def _generator_inner_pool( tool_call_parser_name=self.tool_call_parser_name, reasoning_parser_name=self.reasoning_parser_name, eos_token_ids=self.eos_token_ids, - stop_strings=_request_stop_strings(request), + stop_strings=_request_stop_strings( + request, match_visible=reasoning_parser is not None + ), + include_stop_str_in_output=request.get("include_stop_str_in_output", False), + min_tokens=request.get("min_tokens", 0), guided_decoding=preproc_result.dynamo_preproc.get( "sampling_options", {} ).get("guided_decoding"), @@ -716,6 +751,8 @@ async def _generate_and_stream( created_ts = int(time.time()) stream_interval = self.stream_interval continuous_usage = _continuous_usage_requested(request) + dynamo_stream = None + terminal_items: list[dict[str, Any]] = [] try: dynamo_stream = await self.routed_engine.generate( @@ -799,6 +836,19 @@ async def _generate_and_stream( t_pp0 = time.monotonic() choice = post.process_output(mapped_response) + local_stop = getattr(post, "matched_stop", None) + if local_stop is not None: + finish_reason = "stop" + stop_reason = local_stop + # Workers normally attach usage to their terminal + # chunk. A frontend stop happens before that chunk. + pending_usage = dict(pending_usage or {}) + prompt_tokens = pending_usage.get("prompt_tokens", input_tokens) + pending_usage.update( + prompt_tokens=prompt_tokens, + completion_tokens=cumulative_output_tokens, + total_tokens=prompt_tokens + cumulative_output_tokens, + ) if self.debug_perf: t_pp1 = time.monotonic() @@ -897,16 +947,35 @@ async def _generate_and_stream( dynamo_out["choices"] = [ {**choice, "delta": {}, "logprobs": None} ] - yield {**envelope, "data": payload} - yield {"_dynamo_annotated": True, "data": dynamo_out} + outgoing = [ + {**envelope, "data": payload}, + {"_dynamo_annotated": True, "data": dynamo_out}, + ] else: - yield envelope + outgoing = [envelope] + + if local_stop is not None: + terminal_items = outgoing + break + for item in outgoing: + yield item pending_token_ids = [] pending_log_probs = None pending_top_logprobs = None pending_usage = None first_chunk = False + + # Break releases the async-for iterator reference. Dropping the + # native stream closes its receiver; RoutedEngine signals only + # the child generation context, leaving HTTP completion intact. + if dynamo_stream is not None: + close = getattr(dynamo_stream, "aclose", None) + if close is not None: + await close() + dynamo_stream = None + for item in terminal_items: + yield item except Unknown: raise except Exception as e: @@ -915,6 +984,11 @@ async def _generate_and_stream( f"Error generating response for request {request_id}: {e}" ) from e finally: + if dynamo_stream is not None: + close = getattr(dynamo_stream, "aclose", None) + if close is not None: + await close() + dynamo_stream = None if self.debug_perf and token_count > 0: logger.info( "[perf] sglang stream done: request=%s tokens=%d " diff --git a/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py b/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py new file mode 100644 index 000000000000..7722f2a06ef8 --- /dev/null +++ b/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py @@ -0,0 +1,248 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""CPU-only regression tests using SGLang's actual GLM reasoning parser.""" + +import asyncio +import json + +import pytest +from _routed_engine_fakes import FakeRoutedItem +from sglang.srt.parser.reasoning_parser import ReasoningParser + +from dingo.frontend.sglang_prepost import SglangStreamingPostProcessor +from dingo.frontend.sglang_processor import ( + DEFAULT_SGLANG_MAX_TOKENS, + SglangProcessor, + _build_dynamo_preproc, + _request_stop_strings, +) + +pytestmark = [pytest.mark.unit, pytest.mark.sglang, pytest.mark.gpu_0] + + +class TextTokenizer: + def decode(self, token_ids, *, skip_special_tokens): + return "".join(chr(token_id) for token_id in token_ids) + + +def make_post(stops, **kwargs): + return SglangStreamingPostProcessor( + tokenizer=TextTokenizer(), + tool_call_parser=None, + reasoning_parser=ReasoningParser( + model_type="glm45", stream_reasoning=True, force_reasoning=True + ), + stop_strings=set(stops), + **kwargs, + ) + + +def feed(post, text, finish=None): + return post.process_output( + {"token_ids": list(map(ord, text)), "finish_reason": finish} + ) + + +@pytest.mark.parametrize( + "limits,expected", + [ + ({}, DEFAULT_SGLANG_MAX_TOKENS), + ({"max_tokens": 256}, 256), + ({"max_tokens": 256, "max_completion_tokens": 512}, 512), + ({"max_tokens": 256, "max_completion_tokens": 0}, 0), + ( + {"max_tokens": None, "max_completion_tokens": None}, + DEFAULT_SGLANG_MAX_TOKENS, + ), + ], +) +def test_glm_negative_penalty_is_neutral_with_finite_limit(limits, expected): + result = _build_dynamo_preproc( + {"frequency_penalty": -2, **limits}, [1], "glm-5.3", None + ) + assert result["sampling_options"]["frequency_penalty"] == 0.0 + assert result["stop_conditions"]["max_tokens"] == expected + + +@pytest.mark.parametrize("model", ["glm-5.3", "GLM-5.3", "zai-org/GLM-5.3"]) +@pytest.mark.parametrize("penalty", [-2, -0.5, 0, 0.5, 2]) +def test_glm_frequency_penalty_compatibility(model, penalty): + request = {"model": model, "frequency_penalty": penalty} + result = _build_dynamo_preproc(request, [1], model, None, require_reasoning=True) + assert result["sampling_options"]["frequency_penalty"] == max(0, penalty) + assert result["model"] == model + assert result["require_reasoning"] is True + assert request == {"model": model, "frequency_penalty": penalty} + + +@pytest.mark.parametrize( + "model", ["test-model", "glm-5.2", "glm-5.3-flash", "my-glm-5.3"] +) +def test_other_models_keep_negative_frequency_penalty(model): + result = _build_dynamo_preproc( + {"frequency_penalty": -2}, [1], model, None, reasoning_parser=object() + ) + assert result["sampling_options"]["frequency_penalty"] == -2 + + +def test_message_field_does_not_become_sampling_option(): + result = _build_dynamo_preproc( + {"messages": [{"role": "user", "content": "hi", "frequency_penalty": -2}]}, + [1], + "glm-5.3", + None, + ) + assert result["sampling_options"]["frequency_penalty"] == 0.0 + + +def test_only_string_stops_move_to_reasoning_frontend(): + request = {"stop": ["\n"], "stop_token_ids": [2]} + result = _build_dynamo_preproc( + request, [1], "glm-5.3", [3], reasoning_parser=object() + ) + assert result["stop_conditions"]["stop"] == [] + assert result["stop_conditions"]["stop_token_ids"] == [2] + assert result["eos_token_ids"] == [3] + assert request["stop"] == ["\n"] + plain = _build_dynamo_preproc(request, [1], "glm-5.3", [3]) + assert plain["stop_conditions"]["stop"] == ["\n"] + + +@pytest.mark.parametrize("include", [False, True]) +def test_reasoning_newline_does_not_stop_answer_newline_does(include): + request = {"stop": "\n", "include_stop_str_in_output": include} + post = make_post( + _request_stop_strings(request, match_visible=True), + include_stop_str_in_output=include, + ) + thought = feed(post, "thinking\nmore thinking\n") + assert thought["delta"]["reasoning_content"] == "thinking\nmore thinking\n" + assert thought["finish_reason"] is None + answer = feed(post, "ALPHA\nBETA") + assert answer["delta"]["content"] == ("ALPHA\n" if include else "ALPHA") + assert answer["finish_reason"] == "stop" + assert post.matched_stop == "\n" + assert feed(post, "must not appear") is None + + +@pytest.mark.parametrize("split", range(1, 5)) +def test_stop_across_chunks(split): + post = make_post([""]) + feed(post, "thinking\n") + first = feed(post, "hello"[: 5 + split]) + second = feed(post, ""[split:] + "ignored") + assert first["delta"]["content"] == "hello" + assert first["finish_reason"] is None + assert second["finish_reason"] == "stop" + assert not second["delta"].get("content") + + +def test_partial_stop_flushed_on_length_and_reasoning_only_length(): + post = make_post([""]) + feed(post, "thinking") + feed(post, "helloanswerSTOPdiscardEND") + assert end["delta"]["content"] == "answer" + assert post.matched_stop == "STOP" + post = make_post(["\n"], min_tokens=150) + assert feed(post, "thinkinga\n")["finish_reason"] is None + end = feed(post, "x" * 160 + "\nignored") + assert end["finish_reason"] == "stop" + + +@pytest.mark.parametrize("interval", [1, 20]) +@pytest.mark.parametrize("continuous", [False, True]) +@pytest.mark.parametrize("native_stream", [False, True]) +def test_local_stop_closes_generation_and_emits_terminal_usage( + interval, continuous, native_stream +): + class Engine: + closed = False + consumed = 0 + + async def generate(self, request, **kwargs): + texts = ["thinking\n", "ALPHA\nBETAignored", "never read"] + + # Native AsyncResponseStream has no aclose method: dropping its + # last reference closes the receiver and stops child generation. + engine = self + + class NativeStream: + def __aiter__(self): + return self + + async def __anext__(self): + if engine.consumed == len(texts): + raise StopAsyncIteration + text = texts[engine.consumed] + engine.consumed += 1 + return FakeRoutedItem({"token_ids": list(map(ord, text))}) + + def __del__(self): + engine.closed = True + + if native_stream: + return NativeStream() + + async def stream(): + try: + for text in texts: + self.consumed += 1 + yield FakeRoutedItem({"token_ids": list(map(ord, text))}) + finally: + self.closed = True + + return stream() + + engine = Engine() + processor = SglangProcessor( + tokenizer=TextTokenizer(), + routed_engine=engine, + tool_call_parser_name=None, + reasoning_parser_name="glm45", + eos_token_ids=None, + stream_interval=interval, + ) + + async def collect(): + items = [] + async for item in processor._generate_and_stream( + "test", + { + "model": "glm-5.3", + "nvext": {"extra_fields": ["stop_reason"]}, + "stream_options": {"continuous_usage_stats": continuous}, + }, + {}, + [1, 2], + make_post(["\n"]), + ): + items.append(item) + choices = item.get("data", {}).get("choices", []) + if choices and choices[0]["finish_reason"]: + assert engine.closed + return items + + items = asyncio.run(collect()) + assert engine.consumed == 2 + terminal = items[-1]["data"] + assert terminal["choices"][0]["finish_reason"] == "stop" + assert terminal["choices"][0]["delta"]["content"] == "ALPHA" + generated = len("thinking\nALPHA\nBETAignored") + assert terminal["usage"]["completion_tokens"] == generated + assert terminal["usage"]["total_tokens"] == generated + 2 + assert terminal["nvext"]["stop_reason"] == "\n" + assert json.loads(items[-1]["comment"][0])["output_tokens"] == generated From bc6b4ebec6b19e97a9b719b6d69fba15d1cc1e2e Mon Sep 17 00:00:00 2001 From: guojn1 Date: Mon, 12 Oct 2026 16:19:47 +0800 Subject: [PATCH 2/2] fix(sglang): infer GLM reasoning parser before applying stops Select glm45 for discovered GLM-5.3 and GLM-5.3-Flash thinking templates when no parser is configured. Preserve explicit parser settings and Qwen3 defaults. Cover automatic GLM selection and explicitly configured Qwen3 reasoning stops with regression tests. --- dingo/frontend/README.md | 8 ++ dingo/frontend/sglang_processor.py | 26 ++++ .../tests/test_sglang_reasoning_stops_unit.py | 118 +++++++++++++++++- 3 files changed, 151 insertions(+), 1 deletion(-) diff --git a/dingo/frontend/README.md b/dingo/frontend/README.md index 477e21fcab1c..15883997c8d3 100644 --- a/dingo/frontend/README.md +++ b/dingo/frontend/README.md @@ -31,3 +31,11 @@ applicable). Stops split across chunks are supported; `include_stop_str_in_outpu controls whether the matched string is retained. EOS and `stop_token_ids` remain worker-side conditions. Without a reasoning parser, including requests that set `separate_reasoning: false`, string stops retain their worker-side behavior. + +When no reasoning parser is explicitly configured on the frontend or in worker +metadata, the frontend selects `glm45` for GLM-5.3 / GLM-5.3-Flash if their chat +template opens a thinking block. Qwen3 continues to require explicit `qwen3` +reasoning parser configuration. +Detection uses the discovered model's `config.json`, so serving aliases do not +affect it. Explicit parser settings take precedence. Unknown model types with +thinking templates produce a warning requesting explicit parser configuration. diff --git a/dingo/frontend/sglang_processor.py b/dingo/frontend/sglang_processor.py index d45be5e9db68..32c980909070 100644 --- a/dingo/frontend/sglang_processor.py +++ b/dingo/frontend/sglang_processor.py @@ -246,6 +246,29 @@ def _runtime_config_parser_name( return value if isinstance(value, str) and value else None +def _infer_reasoning_parser_name(source_path: str, chat_template: str | None) -> str | None: + """Infer known think-tag parsers from the discovered model, not its alias.""" + if not detect_force_reasoning_from_template(chat_template): + return None + try: + with open(os.path.join(source_path, "config.json"), encoding="utf-8") as f: + model_config = json.load(f) + except (OSError, ValueError): + logger.warning("Cannot infer reasoning parser from %s/config.json", source_path) + return None + parser = { + "glm_moe_dsa": "glm45", + "glm5_next": "glm45", + }.get(model_config.get("model_type")) + if parser is None: + logger.warning( + "Thinking template detected for model_type=%r without a reasoning parser; " + "configure --dyn-reasoning-parser to separate reasoning and answer stops", + model_config.get("model_type"), + ) + return parser + + def _unsupported_n_message(n: int) -> str: return f"Unsupported value: 'n={n}'. " "This endpoint currently supports only n=1." @@ -1073,6 +1096,9 @@ async def chat_engine_factory( reasoning_parser_name = ( self.reasoning_parser_name or _runtime_config_parser_name(mdc, "reasoning_parser") + or _infer_reasoning_parser_name( + local_dir, getattr(tokenizer, "chat_template", None) + ) ) if tool_call_parser_name: diff --git a/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py b/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py index 7722f2a06ef8..212012f92e15 100644 --- a/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py +++ b/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py @@ -5,22 +5,138 @@ import asyncio import json +from types import SimpleNamespace import pytest -from _routed_engine_fakes import FakeRoutedItem +from _routed_engine_fakes import FakeRoutedEngine, FakeRoutedItem from sglang.srt.parser.reasoning_parser import ReasoningParser +import dingo.frontend.sglang_processor as processor_module from dingo.frontend.sglang_prepost import SglangStreamingPostProcessor from dingo.frontend.sglang_processor import ( DEFAULT_SGLANG_MAX_TOKENS, SglangProcessor, _build_dynamo_preproc, + _infer_reasoning_parser_name, _request_stop_strings, ) pytestmark = [pytest.mark.unit, pytest.mark.sglang, pytest.mark.gpu_0] +@pytest.mark.parametrize( + "model_type,expected", + [ + ("glm_moe_dsa", "glm45"), + ("glm5_next", "glm45"), + ("qwen3", None), + ("qwen3_moe", None), + ("qwen3_next", None), + ("unknown", None), + ], +) +def test_infer_reasoning_parser_requires_known_model_and_thinking_template( + tmp_path, model_type, expected +): + (tmp_path / "config.json").write_text(json.dumps({"model_type": model_type})) + assert ( + _infer_reasoning_parser_name(str(tmp_path), "<|assistant|>") == expected + ) + assert _infer_reasoning_parser_name(str(tmp_path), "plain assistant") is None + + +def test_infer_reasoning_parser_missing_config(tmp_path): + assert _infer_reasoning_parser_name(str(tmp_path), "<|assistant|>") is None + + +@pytest.mark.parametrize( + "model_type,expected,source", + [ + ("glm5_next", "glm45", "inferred"), + ("glm_moe_dsa", "glm45", "inferred"), + ("glm5_next", "glm45", "frontend"), + ("glm5_next", "glm45", "worker"), + ("qwen3", "qwen3", "frontend"), + ("qwen3", "qwen3", "worker"), + ], +) +def test_factory_selects_parser_and_preserves_reasoning_before_stop( + tmp_path, monkeypatch, model_type, expected, source +): + # Exercise discovery -> factory -> parser -> preprocessing/streaming. + # No model weights or tokenizer downloads are needed. + (tmp_path / "config.json").write_text(json.dumps({"model_type": model_type})) + tokenizer = TextTokenizer() + tokenizer.chat_template = "<|assistant|>" + monkeypatch.setattr(processor_module, "_load_tokenizer", lambda *_: tokenizer) + monkeypatch.setattr( + processor_module, "_resolve_model_eos_token_ids", lambda **_: [] + ) + monkeypatch.setattr( + processor_module, "PythonAsyncEngine", lambda generator, _: generator + ) + configured = "qwen3" + selected = expected if source == "inferred" else configured + factory = processor_module.SglangEngineFactory( + SimpleNamespace( + trust_remote_code=False, + preprocess_workers=0, + exclude_tools_when_tool_choice_none=True, + ), + reasoning_parser_name=configured if source == "frontend" else None, + ) + mdc = SimpleNamespace( + model_type=lambda: SimpleNamespace(supports_chat=lambda: True), + local_dir=lambda: str(tmp_path), + name=lambda: "custom-serving-alias", + runtime_config=lambda: ( + {"reasoning_parser": configured} if source == "worker" else {} + ), + ) + engine = FakeRoutedEngine( + [ + {"token_ids": list(map(ord, "thinking\nmore thinking\n"))}, + {"token_ids": list(map(ord, "ALPHA\nBETA"))}, + ] + ) + + async def run(): + generator = await factory.chat_engine_factory(None, mdc, engine) + processor = generator.__self__ + assert processor.reasoning_parser_name == selected + request = {"model": "custom-serving-alias", "stop": ["\n"]} + parser = ReasoningParser( + model_type=processor.reasoning_parser_name, + stream_reasoning=True, + force_reasoning=True, + ) + preproc = _build_dynamo_preproc( + request, [1], request["model"], None, reasoning_parser=parser + ) + assert preproc["stop_conditions"]["stop"] == [] + post = SglangStreamingPostProcessor( + tokenizer=tokenizer, + tool_call_parser=None, + reasoning_parser=parser, + stop_strings={"\n"}, + ) + return [ + x + async for x in processor._generate_and_stream( + "test", request, preproc, [1], post + ) + ] + + items = asyncio.run(run()) + choices = [c for item in items for c in item.get("data", {}).get("choices", [])] + assert ( + "".join(c["delta"].get("reasoning_content", "") for c in choices) + == "thinking\nmore thinking\n" + ) + assert "".join(c["delta"].get("content", "") for c in choices) == "ALPHA" + assert choices[-1]["finish_reason"] == "stop" + + class TextTokenizer: def decode(self, token_ids, *, skip_special_tokens): return "".join(chr(token_id) for token_id in token_ids)