Fix bench_serving non-stream reasoning content (#25298)
This commit is contained in:
@@ -131,6 +131,10 @@ def get_request_headers() -> Dict[str, str]:
|
||||
return headers
|
||||
|
||||
|
||||
def _combine_openai_chat_content(message: Dict[str, Any]) -> str:
|
||||
return (message.get("reasoning_content") or "") + (message.get("content") or "")
|
||||
|
||||
|
||||
def wait_for_endpoint(url: str, timeout_sec: int = 60) -> bool:
|
||||
"""Wait for the server to become ready by polling the given URL."""
|
||||
print(f"Waiting up to {timeout_sec}s for {url} to become ready...")
|
||||
@@ -440,9 +444,8 @@ async def async_request_openai_chat_completions(
|
||||
if args.disable_stream:
|
||||
# Non-streaming response
|
||||
response_json = await response.json()
|
||||
output.generated_text = response_json["choices"][0]["message"][
|
||||
"content"
|
||||
]
|
||||
message = response_json["choices"][0]["message"]
|
||||
output.generated_text = _combine_openai_chat_content(message)
|
||||
output.success = True
|
||||
output.latency = time.perf_counter() - st
|
||||
output.ttft = (
|
||||
@@ -477,9 +480,7 @@ async def async_request_openai_chat_completions(
|
||||
# Reasoning models stream thoughts via
|
||||
# `reasoning_content`; count them like content.
|
||||
delta = choices[0].get("delta") or {}
|
||||
content = (delta.get("reasoning_content") or "") + (
|
||||
delta.get("content") or ""
|
||||
)
|
||||
content = _combine_openai_chat_content(delta)
|
||||
|
||||
if content:
|
||||
timestamp = time.perf_counter()
|
||||
|
||||
@@ -19,6 +19,7 @@ from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
from sglang.bench_serving import (
|
||||
RequestFuncInput,
|
||||
async_request_openai_chat_completions,
|
||||
calculate_metrics,
|
||||
set_global_args,
|
||||
)
|
||||
from sglang.test.ci.ci_register import register_cpu_ci
|
||||
@@ -58,6 +59,24 @@ class _SSEHandler(BaseHTTPRequestHandler):
|
||||
return
|
||||
|
||||
|
||||
class _JSONHandler(BaseHTTPRequestHandler):
|
||||
response_body: dict = {}
|
||||
request_bodies: list = []
|
||||
|
||||
def do_POST(self): # noqa: N802 (BaseHTTPRequestHandler interface)
|
||||
length = int(self.headers.get("Content-Length", "0"))
|
||||
if length:
|
||||
self.request_bodies.append(json.loads(self.rfile.read(length)))
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.end_headers()
|
||||
self.wfile.write(json.dumps(self.response_body).encode())
|
||||
self.wfile.flush()
|
||||
|
||||
def log_message(self, fmt, *args):
|
||||
return
|
||||
|
||||
|
||||
def _make_chunk(content=None, reasoning_content=None, completion_tokens=None):
|
||||
delta = {}
|
||||
if content is not None:
|
||||
@@ -70,6 +89,23 @@ def _make_chunk(content=None, reasoning_content=None, completion_tokens=None):
|
||||
return chunk
|
||||
|
||||
|
||||
def _make_response(content=None, reasoning_content=None, completion_tokens=1):
|
||||
message = {"role": "assistant", "content": content}
|
||||
if reasoning_content is not None:
|
||||
message["reasoning_content"] = reasoning_content
|
||||
return {
|
||||
"choices": [{"index": 0, "message": message, "finish_reason": "length"}],
|
||||
"usage": {"completion_tokens": completion_tokens},
|
||||
}
|
||||
|
||||
|
||||
class _StrictStringTokenizer:
|
||||
def encode(self, text, add_special_tokens=False):
|
||||
if not isinstance(text, str):
|
||||
raise ValueError("text input must be of type `str`")
|
||||
return text.split()
|
||||
|
||||
|
||||
class TestBenchServingReasoningStream(CustomTestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
@@ -206,5 +242,93 @@ class TestBenchServingReasoningStream(CustomTestCase):
|
||||
self.assertGreater(out.ttft, 0.0)
|
||||
|
||||
|
||||
class TestBenchServingReasoningNonStream(CustomTestCase):
|
||||
def _run(self, response_body):
|
||||
set_global_args(
|
||||
Namespace(
|
||||
disable_stream=True,
|
||||
disable_ignore_eos=False,
|
||||
print_requests=False,
|
||||
tokenizer="",
|
||||
header=None,
|
||||
)
|
||||
)
|
||||
port = _free_port()
|
||||
|
||||
class Handler(_JSONHandler):
|
||||
pass
|
||||
|
||||
Handler.response_body = response_body
|
||||
Handler.request_bodies = []
|
||||
server = HTTPServer(("127.0.0.1", port), Handler)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
req = RequestFuncInput(
|
||||
prompt="hello",
|
||||
api_url=f"http://127.0.0.1:{port}/v1/chat/completions",
|
||||
prompt_len=1,
|
||||
output_len=64,
|
||||
model="dummy-model",
|
||||
lora_name="",
|
||||
image_data=None,
|
||||
extra_request_body={},
|
||||
)
|
||||
return (
|
||||
asyncio.run(async_request_openai_chat_completions(req)),
|
||||
Handler.request_bodies,
|
||||
)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
def test_reasoning_only_non_stream_metrics_retokenize_text(self):
|
||||
out, request_bodies = self._run(
|
||||
_make_response(
|
||||
content=None,
|
||||
reasoning_content="Let me think.",
|
||||
completion_tokens=3,
|
||||
)
|
||||
)
|
||||
|
||||
self.assertTrue(out.success, msg=f"request failed: {out.error}")
|
||||
self.assertEqual(out.generated_text, "Let me think.")
|
||||
self.assertEqual(out.output_len, 3)
|
||||
self.assertFalse(request_bodies[0]["stream"])
|
||||
|
||||
metrics, output_lens = calculate_metrics(
|
||||
input_requests=None,
|
||||
outputs=[out],
|
||||
dur_s=1.0,
|
||||
tokenizer=_StrictStringTokenizer(),
|
||||
backend="sglang-oai-chat",
|
||||
)
|
||||
self.assertEqual(metrics.completed, 1)
|
||||
self.assertEqual(output_lens, [3])
|
||||
self.assertEqual(metrics.total_output_retokenized, 3)
|
||||
|
||||
def test_reasoning_then_content_non_stream_accounts_both(self):
|
||||
out, _ = self._run(
|
||||
_make_response(
|
||||
content="answer",
|
||||
reasoning_content="thought ",
|
||||
completion_tokens=2,
|
||||
)
|
||||
)
|
||||
|
||||
self.assertTrue(out.success, msg=f"request failed: {out.error}")
|
||||
self.assertEqual(out.generated_text, "thought answer")
|
||||
self.assertGreater(out.ttft, 0.0)
|
||||
self.assertEqual(out.output_len, 2)
|
||||
|
||||
def test_content_only_non_stream_unchanged(self):
|
||||
out, _ = self._run(_make_response(content="answer", completion_tokens=1))
|
||||
|
||||
self.assertTrue(out.success, msg=f"request failed: {out.error}")
|
||||
self.assertEqual(out.generated_text, "answer")
|
||||
self.assertGreater(out.ttft, 0.0)
|
||||
self.assertEqual(out.output_len, 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user