mirror of
https://github.com/vllm-project/vllm.git
synced 2026-10-28 09:45:26 +08:00
[Benchmark] Add Responses API backend to vllm bench serve (#54628)
Signed-off-by: QHarshil <harshil_c@hotmail.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Chauncey <chaunceyjiang@gmail.com>
This commit is contained in:
5 files changed
+682
-2
No files matched your search
@@ -596,6 +596,46 @@ vllm bench serve \
|
||||
--max-concurrency 512
|
||||
```
|
||||
|
||||
#### Responses API Benchmark
|
||||
|
||||
The `openai-responses` backend benchmarks vLLM's
|
||||
[Responses API](../serving/online_serving/openai_compatible_server.md#responses-api)
|
||||
directly, instead of routing the same workload through `/v1/chat/completions`.
|
||||
|
||||
```bash
|
||||
# Server
|
||||
vllm serve openai/gpt-oss-20b
|
||||
|
||||
# Client
|
||||
vllm bench serve \
|
||||
--backend openai-responses \
|
||||
--endpoint /v1/responses \
|
||||
--model openai/gpt-oss-20b \
|
||||
--dataset-name random \
|
||||
--random-input-len 1024 \
|
||||
--random-output-len 1024 \
|
||||
--num-prompts 200
|
||||
```
|
||||
|
||||
Reasoning deltas (`response.reasoning_text.delta`) count towards TTFT and ITL,
|
||||
because the server is already decoding tokens when it emits them. Only output
|
||||
text (`response.output_text.delta`) is collected as the generated text, which
|
||||
matches how the `openai-chat` backend treats `DeltaMessage.reasoning`. End-to-end
|
||||
latency stops at the last token event, so `latency - ttft` equals `sum(itl)` and
|
||||
TPOT is comparable with the other endpoints. The input and output token counts
|
||||
come from the usage block on `response.completed`, which is still required: a
|
||||
stream that ends without a terminal event is reported as a failed request.
|
||||
|
||||
The backend measures one streamed text-generation request per prompt. Built-in
|
||||
tools, MCP, and multi-turn state via `previous_response_id` are out of scope.
|
||||
All sampling parameter flags are supported except `--min-p`, which the
|
||||
Responses API does not accept.
|
||||
|
||||
!!! warning
|
||||
Do not pass `--extra-body '{"include_reasoning": false}'` when benchmarking
|
||||
a reasoning model. The server still generates reasoning tokens but emits no
|
||||
events for them, so the whole reasoning phase is absorbed into TTFT.
|
||||
|
||||
#### Running With Sampling Parameters
|
||||
|
||||
When using OpenAI-compatible backends such as `vllm`, optional sampling
|
||||
|
||||
@@ -0,0 +1,455 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
||||
"""Tests for the ``openai-responses`` backend of ``vllm bench serve``.
|
||||
|
||||
The Responses stream differs from Chat Completions in ways the benchmark client
|
||||
has to get right: every SSE message is labelled with an ``event:`` line,
|
||||
reasoning and output text arrive as separate token-bearing events, and a
|
||||
request ends with ``response.completed`` rather than with its last token. These
|
||||
tests drive the real request function against a local SSE server so the
|
||||
resulting latency and usage numbers are covered end to end.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from collections.abc import Sequence
|
||||
|
||||
import aiohttp
|
||||
import pytest
|
||||
from aiohttp import web
|
||||
from aiohttp.test_utils import TestServer
|
||||
|
||||
from vllm.benchmarks.lib.endpoint_request_func import (
|
||||
ASYNC_REQUEST_FUNCS,
|
||||
OPENAI_COMPATIBLE_BACKENDS,
|
||||
RequestFuncInput,
|
||||
RequestFuncOutput,
|
||||
async_request_openai_responses,
|
||||
)
|
||||
|
||||
# These tests never touch torch, so skip the GPU cleanup fixture.
|
||||
pytestmark = pytest.mark.skip_global_cleanup
|
||||
|
||||
MODEL = "openai/gpt-oss-20b"
|
||||
|
||||
# Gaps between scripted events. GAP_TOKEN is much larger than GAP_PREFILL so
|
||||
# that a time to first token measured at the wrong event is unambiguous.
|
||||
GAP_PREFILL = 0.15
|
||||
GAP_TOKEN = 0.5
|
||||
GAP_FINALIZE = 0.3
|
||||
|
||||
|
||||
def sse(event_type: str, payload: dict) -> bytes:
|
||||
"""Frame one event the way ``/v1/responses`` does, with an ``event:`` line.
|
||||
|
||||
``ensure_ascii=False`` matches the server, whose pydantic
|
||||
``model_dump_json`` emits non-ASCII characters unescaped.
|
||||
"""
|
||||
data = json.dumps({"type": event_type, **payload}, ensure_ascii=False)
|
||||
return f"event: {event_type}\ndata: {data}\n\n".encode()
|
||||
|
||||
|
||||
def usage_event(
|
||||
input_tokens: int,
|
||||
output_tokens: int,
|
||||
event_type: str = "response.completed",
|
||||
status: str = "completed",
|
||||
):
|
||||
return sse(
|
||||
event_type,
|
||||
{
|
||||
"response": {
|
||||
"status": status,
|
||||
"usage": {
|
||||
"input_tokens": input_tokens,
|
||||
"output_tokens": output_tokens,
|
||||
"total_tokens": input_tokens + output_tokens,
|
||||
},
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def text_delta(delta: str) -> bytes:
|
||||
return sse("response.output_text.delta", {"delta": delta})
|
||||
|
||||
|
||||
def reasoning_delta(delta: str) -> bytes:
|
||||
return sse("response.reasoning_text.delta", {"delta": delta})
|
||||
|
||||
|
||||
# The metadata events the server sends before any token. None of these may
|
||||
# start the TTFT clock or contribute an ITL sample.
|
||||
PREAMBLE = (
|
||||
sse("response.created", {"response": {"status": "in_progress"}})
|
||||
+ sse("response.in_progress", {"response": {"status": "in_progress"}})
|
||||
+ sse("response.output_item.added", {"output_index": 0})
|
||||
+ sse("response.reasoning_part.added", {"output_index": 0})
|
||||
+ sse("response.content_part.added", {"output_index": 0})
|
||||
)
|
||||
|
||||
# The `.done` family the server sends after the last delta. These repeat the
|
||||
# accumulated text, so timing them would add bogus ITL samples and counting
|
||||
# their payload would duplicate the output.
|
||||
DONE_EVENTS = (
|
||||
sse("response.reasoning_text.done", {"text": "The user asks "})
|
||||
+ sse("response.reasoning_part.done", {"output_index": 0})
|
||||
+ sse("response.output_text.done", {"text": "Rayleigh scattering."})
|
||||
+ sse("response.content_part.done", {"output_index": 0})
|
||||
+ sse("response.output_item.done", {"output_index": 0})
|
||||
)
|
||||
|
||||
|
||||
async def _serve(
|
||||
script: Sequence[bytes | float],
|
||||
*,
|
||||
status: int = 200,
|
||||
chunk_size: int | None = None,
|
||||
seen: dict | None = None,
|
||||
**request_kwargs,
|
||||
) -> RequestFuncOutput:
|
||||
"""Run the request function against a server replaying ``script``.
|
||||
|
||||
Args:
|
||||
script: Byte strings to write, interleaved with floats to sleep for.
|
||||
status: HTTP status to respond with. Non-200 skips the script.
|
||||
chunk_size: If set, split every write into chunks of this many bytes to
|
||||
emulate fragmented transport.
|
||||
seen: If given, populated with the request payload.
|
||||
request_kwargs: Extra fields for the ``RequestFuncInput``.
|
||||
|
||||
"""
|
||||
|
||||
async def handler(request: web.Request) -> web.StreamResponse:
|
||||
if seen is not None:
|
||||
seen["payload"] = await request.json()
|
||||
seen["headers"] = dict(request.headers)
|
||||
if status != 200:
|
||||
return web.Response(status=status, reason="Bad Request")
|
||||
|
||||
response = web.StreamResponse(headers={"Content-Type": "text/event-stream"})
|
||||
await response.prepare(request)
|
||||
for item in script:
|
||||
if isinstance(item, float):
|
||||
await asyncio.sleep(item)
|
||||
elif chunk_size is None:
|
||||
await response.write(item)
|
||||
else:
|
||||
for i in range(0, len(item), chunk_size):
|
||||
await response.write(item[i : i + chunk_size])
|
||||
await response.write_eof()
|
||||
return response
|
||||
|
||||
app = web.Application()
|
||||
app.router.add_post("/v1/responses", handler)
|
||||
server = TestServer(app)
|
||||
await server.start_server()
|
||||
try:
|
||||
request_input = RequestFuncInput(
|
||||
prompt="Why is the sky blue?",
|
||||
api_url=str(server.make_url("/v1/responses")),
|
||||
prompt_len=5,
|
||||
output_len=16,
|
||||
model=MODEL,
|
||||
**request_kwargs,
|
||||
)
|
||||
async with aiohttp.ClientSession() as session:
|
||||
return await async_request_openai_responses(request_input, session)
|
||||
finally:
|
||||
await server.close()
|
||||
|
||||
|
||||
def run(script: Sequence[bytes | float], **kwargs) -> RequestFuncOutput:
|
||||
return asyncio.run(_serve(script, **kwargs))
|
||||
|
||||
|
||||
def test_backend_is_registered_as_openai_compatible():
|
||||
assert ASYNC_REQUEST_FUNCS["openai-responses"] is async_request_openai_responses
|
||||
# Membership enables --ignore-eos on random datasets and the sampling
|
||||
# parameter flags, all of which ResponsesRequest accepts.
|
||||
assert "openai-responses" in OPENAI_COMPATIBLE_BACKENDS
|
||||
|
||||
|
||||
def test_request_payload_is_accepted_by_the_server_schema():
|
||||
"""Every field sent must be a real ResponsesRequest field.
|
||||
|
||||
This is what makes the backend safe to list in OPENAI_COMPATIBLE_BACKENDS:
|
||||
the flags that membership enables have to reach the model, and the server
|
||||
drops unknown fields without complaining.
|
||||
"""
|
||||
from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
|
||||
|
||||
sampling = {
|
||||
"temperature": 0.7,
|
||||
"top_p": 0.9,
|
||||
"top_k": 20,
|
||||
"frequency_penalty": 0.2,
|
||||
"presence_penalty": 0.1,
|
||||
"repetition_penalty": 1.05,
|
||||
}
|
||||
seen: dict = {}
|
||||
output = run(
|
||||
[PREAMBLE, text_delta("hi"), usage_event(5, 1)],
|
||||
seen=seen,
|
||||
extra_body=sampling,
|
||||
ignore_eos=True,
|
||||
model_name="served-alias",
|
||||
request_id="bench-7",
|
||||
extra_headers={"x-team": "perf"},
|
||||
)
|
||||
|
||||
assert output.success
|
||||
# --served-model-name wins over --model, as on the other backends.
|
||||
assert seen["payload"]["model"] == "served-alias"
|
||||
# The route is behind validate_json_request, which 415s a missing type.
|
||||
assert seen["headers"]["Content-Type"] == "application/json"
|
||||
assert seen["headers"]["x-request-id"] == "bench-7"
|
||||
assert seen["headers"]["x-team"] == "perf"
|
||||
assert seen["payload"] == {
|
||||
"model": "served-alias",
|
||||
"input": "Why is the sky blue?",
|
||||
"max_output_tokens": 16,
|
||||
"stream": True,
|
||||
"store": False,
|
||||
"ignore_eos": True,
|
||||
**sampling,
|
||||
}
|
||||
assert set(seen["payload"]) <= set(ResponsesRequest.model_fields)
|
||||
ResponsesRequest.model_validate(seen["payload"])
|
||||
|
||||
# `vllm bench serve` rejects --min-p for this backend for the same reason.
|
||||
assert "min_p" not in ResponsesRequest.model_fields
|
||||
|
||||
|
||||
def test_latency_accounting_matches_the_other_endpoints():
|
||||
"""`latency - ttft` must equal `sum(itl)`.
|
||||
|
||||
`serve.py` derives TPOT as `(latency - ttft) / (output_len - 1)`, which only
|
||||
means average inter-token latency when that identity holds. #55508 fixed the
|
||||
same accounting on the chat and audio backends; the trailing events after
|
||||
the last token must not extend the request.
|
||||
"""
|
||||
output = run(
|
||||
[
|
||||
PREAMBLE,
|
||||
GAP_PREFILL / 2,
|
||||
# A keep-alive comment sent during prefill is not a token.
|
||||
b": keep-alive\n\n",
|
||||
GAP_PREFILL / 2,
|
||||
reasoning_delta("The user asks "),
|
||||
GAP_TOKEN,
|
||||
text_delta("Rayleigh"),
|
||||
text_delta(" scattering."),
|
||||
DONE_EVENTS,
|
||||
GAP_FINALIZE,
|
||||
usage_event(5, 8),
|
||||
]
|
||||
)
|
||||
|
||||
assert output.success
|
||||
# Reasoning is timed but not returned, matching how `openai-chat` treats
|
||||
# `DeltaMessage.reasoning`.
|
||||
assert output.generated_text == "Rayleigh scattering."
|
||||
|
||||
# Neither the response.created/in_progress/output_item.added preamble nor
|
||||
# the keep-alive comment may start the clock, and the reasoning delta must,
|
||||
# so TTFT lands on the prefill gap rather than near zero or out at the
|
||||
# first output text delta.
|
||||
assert GAP_PREFILL * 0.8 <= output.ttft < GAP_PREFILL + GAP_TOKEN * 0.5
|
||||
|
||||
# One ITL per token event after the first: the gap before "Rayleigh" and
|
||||
# the back-to-back delta that follows it.
|
||||
assert len(output.itl) == 2
|
||||
assert output.itl[0] >= GAP_TOKEN * 0.5
|
||||
assert output.itl[1] < GAP_TOKEN * 0.5
|
||||
|
||||
# The identity is exact, so no wall-clock margin is involved. The .done
|
||||
# family and the GAP_FINALIZE wait before response.completed sit outside
|
||||
# the request, exactly as the trailing usage chunk does on chat.
|
||||
assert output.latency - output.ttft == pytest.approx(sum(output.itl), abs=1e-9)
|
||||
|
||||
assert output.start_time > 0.0
|
||||
|
||||
|
||||
def test_usage_is_taken_from_the_terminal_event():
|
||||
output = run([PREAMBLE, text_delta("blue"), usage_event(17, 4)])
|
||||
|
||||
assert output.success
|
||||
assert output.prompt_len == 17
|
||||
assert output.output_tokens == 4
|
||||
|
||||
|
||||
@pytest.mark.parametrize("event_type", ["response.completed", "response.incomplete"])
|
||||
def test_incomplete_response_is_a_successful_request(event_type: str):
|
||||
"""Hitting max_output_tokens is the Responses form of finish_reason=length.
|
||||
|
||||
vLLM reports it as `response.completed` carrying `status: "incomplete"`.
|
||||
The spec's dedicated `response.incomplete` event is accepted as well.
|
||||
"""
|
||||
output = run(
|
||||
[
|
||||
PREAMBLE,
|
||||
text_delta("truncated"),
|
||||
usage_event(17, 16, event_type=event_type, status="incomplete"),
|
||||
]
|
||||
)
|
||||
|
||||
assert output.success
|
||||
assert output.generated_text == "truncated"
|
||||
assert output.output_tokens == 16
|
||||
|
||||
|
||||
@pytest.mark.parametrize("chunk_size", [1, 3, 7, 64])
|
||||
def test_output_survives_transport_chunk_boundaries(chunk_size: int):
|
||||
"""Splits inside the event line, the JSON, or a multibyte character."""
|
||||
text = " leading 中 trailing "
|
||||
output = run(
|
||||
[PREAMBLE, text_delta(text), usage_event(5, 6)],
|
||||
chunk_size=chunk_size,
|
||||
)
|
||||
|
||||
assert output.success
|
||||
assert output.generated_text == text
|
||||
assert output.output_tokens == 6
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"char", ["\u2028", "\u2029", "\u0085"], ids=["u2028", "u2029", "u0085"]
|
||||
)
|
||||
def test_unicode_line_separators_do_not_truncate_the_payload(char: str):
|
||||
"""The server emits these unescaped and they are legal inside a JSON string.
|
||||
|
||||
`str.splitlines()` treats them as line breaks, which would cut the `data:`
|
||||
line short and fail the request on valid model output.
|
||||
"""
|
||||
text = f"before{char}after"
|
||||
output = run([PREAMBLE, text_delta(text), usage_event(5, 3)])
|
||||
|
||||
assert output.success, output.error
|
||||
assert output.generated_text == text
|
||||
|
||||
|
||||
def test_empty_data_line_is_skipped():
|
||||
"""A bare `data:` line carries no JSON and must not fail the request."""
|
||||
output = run([PREAMBLE, b"data:\n\n", text_delta("ok"), usage_event(5, 1)])
|
||||
|
||||
assert output.success, output.error
|
||||
assert output.generated_text == "ok"
|
||||
|
||||
|
||||
def test_prompt_len_falls_back_when_usage_is_absent():
|
||||
"""A terminal event without usage keeps the locally computed prompt length."""
|
||||
output = run(
|
||||
[
|
||||
PREAMBLE,
|
||||
text_delta("hi"),
|
||||
sse("response.completed", {"response": {"status": "completed"}}),
|
||||
]
|
||||
)
|
||||
|
||||
assert output.success, output.error
|
||||
assert output.prompt_len == 5
|
||||
|
||||
|
||||
def test_multi_line_data_payload_is_joined_with_newlines():
|
||||
"""The SSE spec allows one payload to span several `data:` lines."""
|
||||
script: list[bytes | float] = [
|
||||
PREAMBLE,
|
||||
b"event: response.output_text.delta\n"
|
||||
b'data: {"type": "response.output_text.delta",\n'
|
||||
b'data: "delta": "ok"}\n\n',
|
||||
usage_event(5, 1),
|
||||
]
|
||||
output = run(script)
|
||||
|
||||
assert output.success, output.error
|
||||
assert output.generated_text == "ok"
|
||||
|
||||
|
||||
def test_done_sentinel_is_skipped():
|
||||
"""Other Responses servers send `[DONE]`, but vLLM does not."""
|
||||
output = run([PREAMBLE, text_delta("ok"), usage_event(5, 1), b"data: [DONE]\n\n"])
|
||||
|
||||
assert output.success, output.error
|
||||
assert output.generated_text == "ok"
|
||||
|
||||
|
||||
def test_error_before_any_token_reports_the_server_error():
|
||||
"""The server's message must win over the generic no-TTFT message."""
|
||||
output = run(
|
||||
[PREAMBLE, b'event: error\ndata: {"type": "error", "message": "boom"}\n\n']
|
||||
)
|
||||
|
||||
assert not output.success
|
||||
assert "boom" in output.error
|
||||
|
||||
|
||||
def test_http_error_is_not_reported_as_success():
|
||||
output = run([], status=400)
|
||||
|
||||
assert not output.success
|
||||
assert output.error
|
||||
|
||||
|
||||
def test_stream_without_tokens_is_not_reported_as_success():
|
||||
"""A stream that finishes without generating anything has no TTFT."""
|
||||
output = run([PREAMBLE, usage_event(5, 0)])
|
||||
|
||||
assert not output.success
|
||||
assert "TTFT" in output.error
|
||||
|
||||
|
||||
def test_truncated_stream_is_not_reported_as_success():
|
||||
"""A stream cut short after a token has no usage and an understated E2EL."""
|
||||
output = run([PREAMBLE, text_delta("partial")])
|
||||
|
||||
assert not output.success
|
||||
assert "terminal event" in output.error
|
||||
|
||||
|
||||
def test_mid_stream_error_is_not_reported_as_success():
|
||||
"""An error payload mid-stream must not be counted as a success.
|
||||
|
||||
vLLM cannot emit this today: its streaming error path fails to validate
|
||||
against the event union, so a generation failure truncates the stream
|
||||
instead. Proxies and other Responses servers do send it.
|
||||
"""
|
||||
output = run(
|
||||
[
|
||||
PREAMBLE,
|
||||
text_delta("partial"),
|
||||
b'event: error\ndata: {"error": {"message": "boom", "code": 500}}\n\n',
|
||||
]
|
||||
)
|
||||
|
||||
assert not output.success
|
||||
assert "boom" in output.error
|
||||
|
||||
|
||||
def test_data_only_framing_is_still_parsed():
|
||||
"""Some proxies drop the ``event:`` line and forward only ``data:``."""
|
||||
script: list[bytes | float] = [
|
||||
b'data: {"type": "response.output_text.delta", "delta": "ok"}\n\n',
|
||||
usage_event(5, 1),
|
||||
]
|
||||
output = run(script)
|
||||
|
||||
assert output.success
|
||||
assert output.generated_text == "ok"
|
||||
|
||||
|
||||
def test_endpoint_path_is_validated():
|
||||
request_input = RequestFuncInput(
|
||||
prompt="hi",
|
||||
api_url="http://localhost:8000/v1/chat/completions",
|
||||
prompt_len=1,
|
||||
output_len=1,
|
||||
model=MODEL,
|
||||
)
|
||||
|
||||
async def call() -> None:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
await async_request_openai_responses(request_input, session)
|
||||
|
||||
with pytest.raises(ValueError, match="responses"):
|
||||
asyncio.run(call())
|
||||
@@ -2148,7 +2148,9 @@ def get_samples(
|
||||
dataset_path=args.dataset_path, disable_shuffle=args.disable_shuffle
|
||||
)
|
||||
# For the "sonnet" dataset, formatting depends on the backend.
|
||||
if args.backend == "openai-chat":
|
||||
# Chat-style backends leave templating to the server; completions
|
||||
# backends need the prompt rendered client side.
|
||||
if args.backend in ("openai-chat", "openai-responses"):
|
||||
input_requests = sonnet_dataset.sample(
|
||||
num_requests=args.num_prompts,
|
||||
input_len=args.sonnet_input_len,
|
||||
|
||||
@@ -447,6 +447,178 @@ async def async_request_openai_chat_completions(
|
||||
return output
|
||||
|
||||
|
||||
def _extract_sse_data(message: str) -> str | None:
|
||||
"""Extract the ``data`` payload of a single SSE message.
|
||||
|
||||
The Responses API stream labels every message with an ``event:`` line
|
||||
before its ``data:`` line, so the payload cannot be recovered by
|
||||
stripping a leading ``data: `` the way the Completions and Chat
|
||||
Completions streams allow.
|
||||
|
||||
Args:
|
||||
message: One complete SSE message, without its trailing blank line.
|
||||
|
||||
Returns:
|
||||
The payload, with multi-line payloads joined by newlines as required
|
||||
by the SSE spec, or None if the message carries no ``data:`` line.
|
||||
|
||||
"""
|
||||
# NOTE: split on "\n" rather than using splitlines(), which also breaks on
|
||||
# U+2028, U+2029 and U+0085. Those are legal inside a JSON string and the
|
||||
# server emits them unescaped, so splitlines() would truncate the payload.
|
||||
data_lines = [
|
||||
line.removeprefix("data:").removeprefix(" ")
|
||||
for line in message.split("\n")
|
||||
if line.startswith("data:")
|
||||
]
|
||||
if not data_lines:
|
||||
return None
|
||||
|
||||
return "\n".join(data_lines)
|
||||
|
||||
|
||||
# Responses events that carry generated tokens. Reasoning deltas are included
|
||||
# because the server is already decoding when it emits them, so excluding them
|
||||
# would report the whole reasoning phase as time to first token.
|
||||
_RESPONSES_TOKEN_DELTA_EVENTS = frozenset(
|
||||
{
|
||||
"response.output_text.delta",
|
||||
"response.reasoning_text.delta",
|
||||
}
|
||||
)
|
||||
|
||||
# Responses events that end a request. Truncation at the output length limit
|
||||
# is a normal benchmark outcome: vLLM reports it as `response.completed` with
|
||||
# `status: "incomplete"`, while the spec also defines a dedicated
|
||||
# `response.incomplete` terminal event.
|
||||
_RESPONSES_TERMINAL_EVENTS = frozenset(
|
||||
{
|
||||
"response.completed",
|
||||
"response.incomplete",
|
||||
}
|
||||
)
|
||||
|
||||
# The spec's failure terminals. vLLM does not emit them today, but
|
||||
# `vllm bench serve` can be pointed at other Responses servers.
|
||||
_RESPONSES_ERROR_EVENTS = frozenset({"error", "response.failed"})
|
||||
|
||||
|
||||
async def async_request_openai_responses(
|
||||
request_func_input: RequestFuncInput,
|
||||
session: aiohttp.ClientSession,
|
||||
pbar: tqdm | None = None,
|
||||
) -> RequestFuncOutput:
|
||||
api_url = request_func_input.api_url
|
||||
_validate_api_url(api_url, "OpenAI Responses API", "responses")
|
||||
|
||||
payload = {
|
||||
"model": request_func_input.model_name
|
||||
if request_func_input.model_name
|
||||
else request_func_input.model,
|
||||
"input": request_func_input.prompt,
|
||||
"max_output_tokens": request_func_input.output_len,
|
||||
"stream": True,
|
||||
# `store` defaults to true. A server started with
|
||||
# VLLM_ENABLE_RESPONSES_API_STORE=1 keeps every stored response for
|
||||
# the lifetime of the process, which would grow its heap over a long
|
||||
# benchmark run.
|
||||
"store": False,
|
||||
}
|
||||
_update_payload_common(payload, request_func_input)
|
||||
|
||||
headers = _get_headers("application/json")
|
||||
_update_headers_common(headers, request_func_input)
|
||||
|
||||
output = RequestFuncOutput()
|
||||
output.prompt_len = request_func_input.prompt_len
|
||||
|
||||
# Only output text is accumulated. Reasoning text is timed but not
|
||||
# returned, matching how `openai-chat` handles `DeltaMessage.reasoning`.
|
||||
generated_text = ""
|
||||
stream_error: str | None = None
|
||||
saw_terminal_event = False
|
||||
first_chunk_received = False
|
||||
st = time.perf_counter()
|
||||
output.start_time = st
|
||||
most_recent_timestamp = st
|
||||
try:
|
||||
async with session.post(url=api_url, json=payload, headers=headers) as response:
|
||||
if response.status == 200:
|
||||
handler = StreamedResponseHandler()
|
||||
|
||||
async for chunk_bytes in response.content.iter_any():
|
||||
for message in handler.add_chunk(chunk_bytes):
|
||||
# NOTE: `_extract_sse_data` returns None for SSE
|
||||
# comments, which servers may send as keep-alives.
|
||||
chunk = _extract_sse_data(message)
|
||||
if not chunk or chunk == "[DONE]":
|
||||
continue
|
||||
|
||||
timestamp = time.perf_counter()
|
||||
data = json.loads(chunk)
|
||||
event_type = data.get("type")
|
||||
|
||||
if event_type in _RESPONSES_TOKEN_DELTA_EVENTS:
|
||||
# First token
|
||||
if not first_chunk_received:
|
||||
first_chunk_received = True
|
||||
output.ttft = timestamp - st
|
||||
|
||||
# Decoding phase
|
||||
else:
|
||||
output.itl.append(timestamp - most_recent_timestamp)
|
||||
|
||||
if event_type == "response.output_text.delta":
|
||||
generated_text += data.get("delta") or ""
|
||||
|
||||
# Only token events advance the end of the
|
||||
# request, so `latency - ttft` stays equal to
|
||||
# `sum(itl)` and TPOT means what serve.py says.
|
||||
most_recent_timestamp = timestamp
|
||||
elif event_type in _RESPONSES_TERMINAL_EVENTS:
|
||||
saw_terminal_event = True
|
||||
usage = (data.get("response") or {}).get("usage") or {}
|
||||
if (n_output := usage.get("output_tokens")) is not None:
|
||||
output.output_tokens = n_output
|
||||
if (n_input := usage.get("input_tokens")) is not None:
|
||||
output.prompt_len = n_input
|
||||
elif event_type in _RESPONSES_ERROR_EVENTS or data.get("error"):
|
||||
stream_error = json.dumps(data.get("error") or data)
|
||||
|
||||
output.generated_text = generated_text
|
||||
output.latency = most_recent_timestamp - st
|
||||
if stream_error is not None:
|
||||
output.success = False
|
||||
output.error = stream_error
|
||||
elif not first_chunk_received:
|
||||
output.success = False
|
||||
output.error = (
|
||||
"Never received a token delta to calculate TTFT."
|
||||
"This response will be marked as failed!"
|
||||
)
|
||||
elif not saw_terminal_event:
|
||||
# Without the terminal event the response is truncated and
|
||||
# its usage block is missing.
|
||||
output.success = False
|
||||
output.error = (
|
||||
"Stream ended without a terminal event."
|
||||
"This response will be marked as failed!"
|
||||
)
|
||||
else:
|
||||
output.success = True
|
||||
else:
|
||||
output.error = response.reason or ""
|
||||
output.success = False
|
||||
except Exception:
|
||||
output.success = False
|
||||
exc_info = sys.exc_info()
|
||||
output.error = "".join(traceback.format_exception(*exc_info))
|
||||
|
||||
if pbar:
|
||||
pbar.update(1)
|
||||
return output
|
||||
|
||||
|
||||
async def async_request_openai_audio(
|
||||
request_func_input: RequestFuncInput,
|
||||
session: aiohttp.ClientSession,
|
||||
@@ -920,6 +1092,7 @@ ASYNC_REQUEST_FUNCS: dict[str, RequestFunc] = {
|
||||
"vllm": async_request_openai_completions,
|
||||
"openai": async_request_openai_completions,
|
||||
"openai-chat": async_request_openai_chat_completions,
|
||||
"openai-responses": async_request_openai_responses,
|
||||
"openai-audio": async_request_openai_audio,
|
||||
"openai-embeddings": async_request_openai_embeddings,
|
||||
"openai-embeddings-chat": async_request_openai_embeddings_chat,
|
||||
@@ -947,5 +1120,10 @@ POOLING_BACKENDS = {
|
||||
OPENAI_COMPATIBLE_BACKENDS = [
|
||||
k
|
||||
for k, v in ASYNC_REQUEST_FUNCS.items()
|
||||
if v in (async_request_openai_completions, async_request_openai_chat_completions)
|
||||
if v
|
||||
in (
|
||||
async_request_openai_completions,
|
||||
async_request_openai_chat_completions,
|
||||
async_request_openai_responses,
|
||||
)
|
||||
]
|
||||
@@ -2216,6 +2216,11 @@ async def main_async(args: argparse.Namespace) -> dict[str, Any]:
|
||||
"Sampling parameters are only supported by openai-compatible backends."
|
||||
)
|
||||
|
||||
# The Responses API accepts every sampling parameter above except
|
||||
# min_p, which it would silently drop as an unknown field.
|
||||
if args.backend == "openai-responses" and "min_p" in sampling_params:
|
||||
raise ValueError("--min-p is not supported by the Responses API.")
|
||||
|
||||
if "temperature" not in sampling_params:
|
||||
print(
|
||||
"WARNING: vllm bench serve no longer sets temperature==0 (greedy) "
|
||||
|
||||
Reference in new issue
Block a user