From 155a9e72379790d902afdcfa0f0b93d1d92de87d Mon Sep 17 00:00:00 2001 From: Lianmin Zheng Date: Sat, 29 Nov 2025 13:56:15 -0800 Subject: [PATCH] Fix condition for streaming output_ids in tokenizer manager (#13759) Signed-off-by: Xinyuan Tong Co-authored-by: Chang Su Co-authored-by: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com> Co-authored-by: Xinyuan Tong --- python/sglang/bench_one_batch_server.py | 2 + python/sglang/srt/entrypoints/context.py | 36 ++++++++-------- .../sglang/srt/entrypoints/harmony_utils.py | 6 +-- .../srt/entrypoints/openai/serving_chat.py | 15 +++---- .../sglang/srt/managers/tokenizer_manager.py | 2 +- test/srt/test_gpt_oss_common.py | 41 +++++++++++++++++++ 6 files changed, 72 insertions(+), 30 deletions(-) diff --git a/python/sglang/bench_one_batch_server.py b/python/sglang/bench_one_batch_server.py index 0176cc292..159a3a935 100644 --- a/python/sglang/bench_one_batch_server.py +++ b/python/sglang/bench_one_batch_server.py @@ -578,6 +578,8 @@ def run_benchmark(server_args: ServerArgs, bench_args: BenchArgs): if is_in_ci() and bench_args.append_to_github_summary: write_github_step_summary(summary) + else: + print(summary) # Save results as pydantic models in the JSON format if bench_args.pydantic_result_filename: diff --git a/python/sglang/srt/entrypoints/context.py b/python/sglang/srt/entrypoints/context.py index 972c0f4f3..083e75f17 100644 --- a/python/sglang/srt/entrypoints/context.py +++ b/python/sglang/srt/entrypoints/context.py @@ -84,14 +84,6 @@ class HarmonyContext(ConversationContext): if isinstance(output, dict) and "output_ids" in output: output_token_ids = output["output_ids"] - # TODO: REMOVE here: - # Very hacky, find the first occurrence of token 200006 and cut from there - try: - start_index = output_token_ids.index(200006) - output_token_ids = output_token_ids[start_index:] - except ValueError: - pass - for token_id in output_token_ids: self.parser.process(token_id) output_msgs = self.parser.messages @@ -189,6 +181,7 @@ class StreamingHarmonyContext(HarmonyContext): self.parser = get_streamable_parser_for_assistant() self.encoding = get_encoding() self.last_tok = None + self.num_processed_tokens = 0 @property def messages(self) -> list: @@ -199,16 +192,25 @@ class StreamingHarmonyContext(HarmonyContext): # RequestOutput from SGLang with outputs output_token_ids = output["output_ids"] - # TODO: REMOVE here: - # Very hacky, find the first occurrence of token 200006 and cut from there - # Find the first occurrence of token 200006 and cut from there - try: - start_index = output_token_ids.index(200006) - output_token_ids = output_token_ids[start_index:] - except ValueError: - pass + # Check if we need to handle cumulative tokens + meta_info = output.get("meta_info", {}) + completion_tokens = meta_info.get("completion_tokens") + if ( + completion_tokens is not None + and len(output_token_ids) == completion_tokens + ): + # Case 1: When --stream-output is not set. + # The output_ids contains all tokens generated so far. + # We only need to process the new tokens. + new_token_ids = output_token_ids[self.num_processed_tokens :] + self.num_processed_tokens = len(output_token_ids) + else: + # Case 2: When --stream-output is set. + # The output_ids contains only the new tokens. + new_token_ids = output_token_ids + self.num_processed_tokens += len(output_token_ids) - for token_id in output_token_ids: + for token_id in new_token_ids: self.parser.process(token_id) else: diff --git a/python/sglang/srt/entrypoints/harmony_utils.py b/python/sglang/srt/entrypoints/harmony_utils.py index 68bbbf094..86d4356dc 100644 --- a/python/sglang/srt/entrypoints/harmony_utils.py +++ b/python/sglang/srt/entrypoints/harmony_utils.py @@ -345,11 +345,7 @@ def parse_remaining_state(parser: StreamableParser): return [reasoning_item] elif parser.current_channel == "final": output_text = ResponseOutputText( - content=[ - ResponseReasoningTextContent( - text=parser.current_content, type="reasoning_text" - ) - ], + text=parser.current_content, annotations=[], # TODO type="output_text", logprobs=None, # TODO diff --git a/python/sglang/srt/entrypoints/openai/serving_chat.py b/python/sglang/srt/entrypoints/openai/serving_chat.py index b6880e5e8..77903c5bb 100644 --- a/python/sglang/srt/entrypoints/openai/serving_chat.py +++ b/python/sglang/srt/entrypoints/openai/serving_chat.py @@ -75,6 +75,13 @@ class OpenAIServingChat(OpenAIServingBase): f"Using default chat sampling params from model generation config: {self.default_sampling_params}", ) + # Check if the model is a GPT-OSS model + self.is_gpt_oss = ( + hasattr(self.tokenizer_manager.model_config, "hf_config") + and hasattr(self.tokenizer_manager.model_config.hf_config, "model_type") + and self.tokenizer_manager.model_config.hf_config.model_type == "gpt_oss" + ) + def _request_id_prefix(self) -> str: return "chatcmpl-" @@ -205,14 +212,8 @@ class OpenAIServingChat(OpenAIServingBase): self, request: ChatCompletionRequest, is_multimodal: bool ) -> MessageProcessingResult: """Process chat messages and apply chat template""" - is_gpt_oss = ( - hasattr(self.tokenizer_manager.model_config, "hf_config") - and hasattr(self.tokenizer_manager.model_config.hf_config, "model_type") - and self.tokenizer_manager.model_config.hf_config.model_type == "gpt_oss" - ) - # GptOss model needs to keep special tokens for harmony parsing - if is_gpt_oss: + if self.is_gpt_oss: request.skip_special_tokens = False tool_call_constraint = None diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 51602e3f1..b90cf0616 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -1623,7 +1623,7 @@ class TokenizerManager(TokenizerCommunicatorMixin): if isinstance(recv_obj, BatchStrOutput): state.text += recv_obj.output_strs[i] - if state.obj.stream: + if self.server_args.stream_output and state.obj.stream: state.output_ids.extend(recv_obj.output_ids[i]) output_token_ids = state.output_ids[state.last_output_offset :] state.last_output_offset = len(state.output_ids) diff --git a/test/srt/test_gpt_oss_common.py b/test/srt/test_gpt_oss_common.py index 6be739277..68402b5e0 100644 --- a/test/srt/test_gpt_oss_common.py +++ b/test/srt/test_gpt_oss_common.py @@ -1,8 +1,11 @@ +import json import os from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace from typing import Dict, List, Literal, Optional +import requests + from sglang.srt.utils import is_hip, kill_process_tree from sglang.test.run_eval import run_eval from sglang.test.test_utils import ( @@ -60,6 +63,8 @@ class BaseTestGptOss(CustomTestCase): ) try: + self._check_streaming_responses_api_request(model) + # run multiple tests in parallel since we are mostly bound by the longest generate sequence # instead of the number of questions with ThreadPoolExecutor(max_workers=4) as executor: @@ -79,6 +84,42 @@ class BaseTestGptOss(CustomTestCase): finally: kill_process_tree(process.pid) + def _check_streaming_responses_api_request(self, model): + # Use requests to verify /v1/responses streaming + url = f"{_base_url}/v1/responses" + payload = { + "model": model, + "input": "What is 1 + 1?", + "stream": True, + "temperature": 0, + } + + response = requests.post(url, json=payload, stream=True) + if response.status_code != 200: + print(f"Response API failed: {response.text}") + response.raise_for_status() + + content = "" + for line in response.iter_lines(): + if line: + decoded_line = line.decode("utf-8") + if decoded_line.startswith("data: "): + data_str = decoded_line[6:] + if data_str.strip() == "[DONE]": + break + + try: + data = json.loads(data_str) + if data.get("type") == "response.output_text.delta": + delta = data.get("delta", "") + content += delta + except json.JSONDecodeError: + pass + + print(f"Streaming check response: {content}") + self.assertTrue(len(content) > 0) + self.assertIn("2", content) + def _run_one_eval(self, model, reasoning_effort, expected_score): args = SimpleNamespace( base_url=_base_url,