Fix condition for streaming output_ids in tokenizer manager (#13759)

Signed-off-by: Xinyuan Tong <xinyuantong.cs@gmail.com>
Co-authored-by: Chang Su <chang.s.su\n@oracle.com>
Co-authored-by: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com>
Co-authored-by: Xinyuan Tong <xinyuantong.cs@gmail.com>

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:

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:

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

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

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)

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,