iHore/vLLM
8
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

[Frontend] Add output_mode to /inference/v1/generate (RFC #56851 Phase 1) (#58588)

Browse Source 
Signed-off-by: Martin Hickey <martin.hickey@ie.ibm.com>
This commit is contained in:
Martin Hickey
2026-10-03 19:38:25 +08:00
committed by GitHub
parent 5f30fc7031
commit bc21cba967
18 changed files with 1356 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/features/disagg_prefill.md
+6
View File
@@ -79,6 +79,12 @@ decode = client.chat.completions.create(
If `messages` has non-text content or `echo` is set, the ids are ignored and `messages` is rendered instead, so it must match the prefill request. Otherwise `kv_transfer_params["prompt_token_ids"]` must be a non-empty list of non-negative integers, or the request fails with HTTP 400, as it always does on `/v1/chat/completions/batch`.
## Generate API output modes
When the prefill and decode stages use the [Generate API](../serving/online_serving/token_in_token_out.md) (`/inference/v1/generate`), only the decode response reaches the client, so set `output_mode` on the decode request only. A proxy that reuses the client's request body for the prefill request must reset `output_mode` to `tokens` there. A prefill instance started with `--tokens-only` has no tokenizer and rejects `output_mode: "text"` with a 400.
A decode instance that returns text needs a tokenizer, so start it with `--enable-scale-out` and without `--tokens-only`. Prefill instances can keep `--tokens-only`.
## Development
We implement disaggregated prefilling by running 2 vLLM instances. One for prefill (we call it prefill instance) and one for decode (we call it decode instance), and then use a connector to transfer the prefill KV caches and results from prefill instance to decode instance.
docs/serving/online_serving/README.md
+4 -2
View File
@@ -125,8 +125,10 @@ Scale-out APIs are disabled by default on `vllm serve`. Set `--enable-scale-out`
### Tokens IN <> Tokens OUT APIs
- `/inference/v1/generate` - Generate completions
- `/abort_requests` - Abort in-flight requests (only when `--tokens-only` is also set)
- [Generate API](token_in_token_out.md) (`/inference/v1/generate`)
- Generate completions from token IDs, optionally with detokenized text (`output_mode`)
- `/inference/v1/abort_requests` - Abort in-flight requests (registered wherever `/inference/v1/generate` is)
- `/abort_requests` - Unauthenticated alias of `/inference/v1/abort_requests` (`--tokens-only` only)
### Renderer APIs
docs/serving/online_serving/derenderer.md
+3 -1
View File
@@ -40,6 +40,8 @@ Seeding only reads the last few IDs (7 today), so a suffix of the prompt is enou
Each request wraps the engine's `GenerateResponse`(s) together with the caller metadata needed to reconstruct the final response without a GPU.
Only `output_mode: "tokens"` responses are accepted which is the default. A response from a request with `output_mode: "text"` gets a 400 because its text is already detokenized with stop strings removed. Responses without `output_mode`, from older servers, are treated as `tokens`.
`/v1/chat/completions/derender`:
??? code
@@ -127,7 +129,7 @@ Streaming derender is stateless on the server side: all mutable state lives in t
- `logprob_context_token_ids`: the trailing sampled token IDs (at most 4) from previous chunks, used to seed byte-fallback (U+FFFD) correction so multi-byte characters whose tokens split across chunk boundaries still resolve to real strings
- `logprob_text_offset`: the cumulative emitted text length, so `text_offset` in completion streaming logprobs stays absolute across chunks instead of restarting at 0
When a streamed `GenerateResponseStreamChoice` carries `logprobs`, the `token_id:N` placeholders are resolved per chunk and the resolved logprobs are attached to the corresponding streamed choice — for chat as `ChatCompletionLogProbs`, for completions converted to the flat `CompletionLogProbs` lists. Chunks without `logprobs` produce choices with `logprobs: null`.
When a streamed `GenerateTokensStreamChoice` carries `logprobs`, the `token_id:N` placeholders are resolved per chunk and the resolved logprobs are attached to the corresponding streamed choice. For chat it is as `ChatCompletionLogProbs` and for completions it is converted to the flat `CompletionLogProbs` lists. Chunks without `logprobs` produce choices with `logprobs: null`.
## Example
docs/serving/online_serving/token_in_token_out.md
+71
View File
@@ -0,0 +1,71 @@
# Tokens In <> Tokens Out API
`/inference/v1/generate` takes prompt token IDs and returns generated token IDs. It is the generation step of the `render → generate → derender` pipeline used in disaggregated serving (see the [Renderer APIs](renderer.md) and [Derenderer APIs](derenderer.md)).
The endpoint is registered on `vllm serve` with `--enable-scale-out`, and always with `--tokens-only`. `--tokens-only` also skips loading the tokenizer, so the server can't turn token IDs back into text.
## Output modes
The request field `output_mode` selects how much of the postprocessing the server does for the caller.
| `output_mode` | Response adds | Server needs |
| --- | --- | --- |
| `tokens` (default) | nothing, token IDs only | nothing |
| `text` | `text` on every choice and decoded tokens with `bytes` in `logprobs` | a tokenizer |
`text` returns text and token IDs in a single call, without a separate [derender](derenderer.md) hop per chunk. This suits latency sensitive streaming, where the derender hop lands on every output token.
Every response and every stream chunk echoes `output_mode`. Servers that predate the field ignore it and return token IDs only with a 200, so check that the `output_mode` in the response matches the one you sent. Older servers don't return it.
```python
import httpx
response = httpx.post(
"http://localhost:8000/inference/v1/generate",
json={
"token_ids": [151644, 872, 198],
"sampling_params": {"max_tokens": 32},
"output_mode": "text",
},
).json()
assert response["output_mode"] == "text"
print(response["choices"][0]["text"])
print(response["choices"][0]["token_ids"])
```
### Text
- `text` comes from the request's own detokenizer, so `skip_special_tokens`, `spaces_between_special_tokens`, `stop` and `include_stop_str_in_output` behave as they do on `/v1/completions`.
- A matched stop string is cut from `text` unless `include_stop_str_in_output` is set. `token_ids` keeps every generated token, so decoding `token_ids` yourself brings the stop string back.
- `/derender` only accepts `output_mode: "tokens"` responses and returns a 400 for text responses which are already detokenized.
- With `stream: true`, each choice's `text` is the delta since the previous chunk. A chunk is sent whenever the engine output carries new text or a `finish_reason`, even with no new token IDs. That covers text held back for stop string matching and the final output after an abort.
### Logprobs
With `output_mode: "tokens"`, logprob entries carry `token_id:N` placeholders. With `output_mode: "text"`, they carry the decoded token strings and their UTF-8 `bytes` for the sampled token and every entry in `top_logprobs`. A server started with `--return-tokens-as-token-ids` always returns the placeholders, as `/v1/completions` does.
### Errors
The server returns a 400 for:
- `output_mode: "text"` on a server without a tokenizer (`--tokens-only` or `--skip-tokenizer-init`).
- `output_mode: "text"` with `sampling_params.detokenize: false`.
- An unsupported `output_mode` value.
For using `output_mode` with separate prefill and decode pools, see [Disaggregated Prefilling](../../features/disagg_prefill.md#generate-api-output-modes).
## Aborting requests
`POST /inference/v1/abort_requests` aborts in-flight requests. It is registered wherever `/inference/v1/generate` is and requires the API key when `--api-key` is set, like `/inference/v1/generate`.
```bash
curl -X POST http://localhost:8000/inference/v1/abort_requests \
-H "Authorization: Bearer $VLLM_API_KEY" \
-H "Content-Type: application/json" \
-d '{"request_ids": ["generate-tokens-42"]}'
```
The request ID is the `request_id` the server returned: the `X-Request-Id` header or body `request_id` you sent, prefixed with `generate-tokens-`. A missing `request_ids` returns a 400. The response is empty and the abort finishes in the background.
With `--tokens-only`, the same handler is also served at `POST /abort_requests` for existing deployments. That path doesn't require the API key, even when `--api-key` is set. See [Security](../../usage/security.md#unprotected-endpoints-no-api-key-required).
docs/usage/security.md
+1 -1
View File
@@ -206,7 +206,7 @@ The following endpoints **do not require authentication** even when `--api-key`
- `/init_weight_transfer_engine` - Initialize weight transfer engine for RLHF
- `/update_weights` - Update model weights (can alter model behavior)
- `/get_world_size` - Get distributed world size
- `/abort_requests` - Abort in-flight requests (available with `--tokens-only`)
- `/abort_requests` - Abort in-flight requests (available with `--tokens-only`. Use the authenticated `/inference/v1/abort_requests` otherwise)
**Utility endpoints:**
tests/entrypoints/scale_out/derender/test_derender_parity.py
+242
View File
@@ -22,6 +22,10 @@ The streaming cases do the same for `stream=true`. Each coupled chunk's
`token_ids` becomes one `/inference/v1/generate` stream chunk, fed
through the chunked derender path with `stream_state` threaded across calls
the way a client would, so both parsers see the same chunk boundaries.
The text level cases pin `/inference/v1/generate` with `output_mode="text"`
against `/v1/completions/derender` on the same token IDs, batch and streaming,
plus logprob resolution and stop string handling.
"""
import json
@@ -440,3 +444,241 @@ async def test_stream_parity_reasoning_and_tool_call(client, chunking):
assert any(len(ch["token_ids"]) > 1 for ch in coupled_choices)
if not (coupled["reasoning"] and coupled["tool_calls"]):
pytest.skip("Model did not emit both a <think> block and a tool call")
# ---------------------------------------------------------------------------
# Inline text level: generate(output_mode="text") == derender(generate(...))
# ---------------------------------------------------------------------------
TEXT_MESSAGES = [
{"role": "user", "content": "What is 2+2? Answer in one short sentence."}
]
async def _render_token_ids(
client: httpx.AsyncClient, messages: list[dict]
) -> list[int]:
resp = await client.post(
"/v1/chat/completions/render", json={"model": MODEL, "messages": messages}
)
assert resp.status_code == 200, resp.text
return resp.json()["token_ids"]
def _generate_payload(
token_ids: list[int], output_mode: str, stream: bool = False, **sampling
) -> dict:
return {
"model": MODEL,
"token_ids": token_ids,
"sampling_params": {"temperature": 0, "max_tokens": 128, **sampling},
"output_mode": output_mode,
"stream": stream,
}
async def _generate(
client: httpx.AsyncClient, token_ids: list[int], output_mode: str, **sampling
) -> dict:
resp = await client.post(
"/inference/v1/generate",
json=_generate_payload(token_ids, output_mode, **sampling),
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["output_mode"] == output_mode
return data
async def _generate_text_stream(
client: httpx.AsyncClient, token_ids: list[int], **sampling
) -> list[dict]:
"""Stream generate at the text level and return the choice of every chunk
that has one."""
choices: list[dict] = []
async with client.stream(
"POST",
"/inference/v1/generate",
json=_generate_payload(token_ids, "text", stream=True, **sampling),
) as resp:
assert resp.status_code == 200, await resp.aread()
async for line in resp.aiter_lines():
if not line.startswith("data: ") or line == "data: [DONE]":
continue
chunk = json.loads(line[len("data: ") :])
assert "error" not in chunk, chunk
assert chunk["output_mode"] == "text"
choices.extend(chunk["choices"])
return choices
async def _derender_completion_text(
client: httpx.AsyncClient, output_ids: list[int], finish_reason: str
) -> str:
resp = await client.post(
"/v1/completions/derender",
json={
"model": MODEL,
"generate_responses": [
{
"request_id": "text-parity",
"choices": [
{
"index": 0,
"token_ids": output_ids,
"finish_reason": finish_reason,
}
],
}
],
},
)
assert resp.status_code == 200, resp.text
return resp.json()["choices"][0]["text"]
async def _derender_completion_stream_text(
client: httpx.AsyncClient, choices: list[dict]
) -> str:
"""Replay each generate chunk's token IDs through the streaming
completions derender endpoint with the same chunk boundaries."""
state = None
text = ""
for choice in choices:
resp = await client.post(
"/v1/completions/derender",
json={
"stream": True,
"model": MODEL,
"generate_chunk": {
"request_id": "text-parity-stream",
"choices": [
{
"index": 0,
"token_ids": choice.get("token_ids") or [],
"finish_reason": choice.get("finish_reason"),
}
],
},
"stream_state": state,
},
)
assert resp.status_code == 200, resp.text
data = resp.json()
state = data["stream_state"]
text += "".join(ch["text"] for ch in data["chunk"]["choices"])
return text
@pytest.mark.asyncio
async def test_text_parity_batch(client):
"""Inline text matches /v1/completions/derender on the same token IDs."""
token_ids = await _render_token_ids(client, TEXT_MESSAGES)
inline = (await _generate(client, token_ids, "text"))["choices"][0]
derendered = await _derender_completion_text(
client, inline["token_ids"], inline["finish_reason"]
)
assert inline["text"]
assert inline["text"] == derendered
@pytest.mark.asyncio
async def test_text_parity_stream(client):
"""Streamed inline text matches both the streaming and the non-streaming
derender of the same token IDs."""
token_ids = await _render_token_ids(client, TEXT_MESSAGES)
choices = await _generate_text_stream(client, token_ids)
inline = "".join(ch["text"] for ch in choices)
output_ids = [t for ch in choices for t in ch.get("token_ids") or []]
finish_reason = choices[-1]["finish_reason"]
assert finish_reason is not None
assert inline == await _derender_completion_stream_text(client, choices)
assert inline == await _derender_completion_text(client, output_ids, finish_reason)
@pytest.mark.asyncio
async def test_text_parity_logprobs(client):
"""Inline logprob tokens and bytes match /derender resolving the
token_id:N placeholders of the tokens level."""
token_ids = await _render_token_ids(client, TEXT_MESSAGES)
sampling = {"logprobs": 3}
tokens_choice = (await _generate(client, token_ids, "tokens", **sampling))[
"choices"
][0]
text_choice = (await _generate(client, token_ids, "text", **sampling))["choices"][0]
assert text_choice["token_ids"] == tokens_choice["token_ids"], (
"greedy (temperature=0) generation was expected to be deterministic "
"across the two generate calls"
)
disagg = await _disagg(
client,
tokens_choice["token_ids"],
len(token_ids),
tokens_choice["finish_reason"],
{"model": MODEL, "messages": TEXT_MESSAGES},
logprobs=tokens_choice["logprobs"],
)
inline_content = text_choice["logprobs"]["content"]
derender_content = disagg["choices"][0]["logprobs"]["content"]
assert len(inline_content) == len(derender_content)
for inline_entry, derender_entry in zip(inline_content, derender_content):
assert inline_entry["token"] == derender_entry["token"]
assert inline_entry["bytes"] == derender_entry["bytes"]
assert [
(top["token"], top["bytes"]) for top in inline_entry["top_logprobs"]
] == [(top["token"], top["bytes"]) for top in derender_entry["top_logprobs"]]
@pytest.mark.asyncio
@pytest.mark.parametrize("stream", [False, True])
async def test_text_parity_stop_string(client, stream):
"""Inline text follows the engine's stop handling, like /v1/completions.
token_ids keep every generated token, so /derender brings the matched
stop string back. Its text must start with the inline text and only
differ after it.
"""
stop = "\n"
token_ids = await _render_token_ids(
client,
[{"role": "user", "content": "Write two short lines about the sea."}],
)
resp = await client.post(
"/v1/completions",
json={
"model": MODEL,
"prompt": token_ids,
"temperature": 0,
"max_tokens": 128,
"stop": [stop],
},
)
assert resp.status_code == 200, resp.text
coupled = resp.json()["choices"][0]
if stream:
choices = await _generate_text_stream(client, token_ids, stop=[stop])
inline_text = "".join(ch["text"] for ch in choices)
output_ids = [t for ch in choices for t in ch.get("token_ids") or []]
finish_reason = choices[-1]["finish_reason"]
else:
inline = (await _generate(client, token_ids, "text", stop=[stop]))["choices"][0]
inline_text = inline["text"]
output_ids = inline["token_ids"]
finish_reason = inline["finish_reason"]
assert inline_text == coupled["text"]
assert finish_reason == coupled["finish_reason"]
assert stop not in inline_text
derendered = await _derender_completion_text(client, output_ids, finish_reason)
assert derendered.startswith(inline_text)
if coupled["stop_reason"] != stop:
pytest.skip("Model did not emit the stop string")
assert derendered[len(inline_text) :].startswith(stop)
tests/entrypoints/scale_out/derender/test_derender_stream.py
+23 -22
View File
@@ -50,10 +50,11 @@ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
DerenderCompletionRequest,
DerenderCompletionStreamRequest,
DerenderStreamState,
GenerateResponse,
GenerateResponseChoice,
GenerateResponseStreamChoice,
GenerateStreamResponse,
GenerateTokensChoice,
GenerateTokensResponse,
GenerateTokensStreamChoice,
GenerateTokensStreamResponse,
)
from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo
from vllm.parser import Parser
@@ -195,10 +196,10 @@ def _make_stream_chunk(
prompt_token_ids: list[int] | None = None,
) -> GenerateStreamResponse:
"""Build a GenerateStreamResponse SSE chunk."""
return GenerateStreamResponse(
return GenerateTokensStreamResponse(
request_id=request_id,
choices=[
GenerateResponseStreamChoice(
GenerateTokensStreamChoice(
index=index,
token_ids=token_ids,
finish_reason=finish_reason,
@@ -233,7 +234,7 @@ def _make_usage_chunk(
request_id: str = "test-req",
) -> GenerateStreamResponse:
"""Build a usage only final SSE chunk (empty choices)."""
return GenerateStreamResponse(
return GenerateTokensStreamResponse(
request_id=request_id,
choices=[],
usage=UsageInfo(
@@ -326,11 +327,11 @@ def test_non_streaming_derender_initializes_parser_prefix(
)
monkeypatch.setattr(derenderer, "parser", MagicMock(return_value=parser))
generated_ids = tokenizer.encode("answer", add_special_tokens=False)
response = GenerateResponse(
response = GenerateTokensResponse(
request_id="test",
prompt_token_ids=[11, 12],
choices=[
GenerateResponseChoice(
GenerateTokensChoice(
index=0,
token_ids=generated_ids,
finish_reason="length",
@@ -583,10 +584,10 @@ class TestPromptSeededLeadingSpace:
):
prompt_ids, output_ids, expected = leading_space_ids
choices = await derenderer.derender_chat(
GenerateResponse(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateResponseChoice(
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
@@ -601,10 +602,10 @@ class TestPromptSeededLeadingSpace:
):
prompt_ids, output_ids, expected = leading_space_ids
choices = await derenderer.derender_chat(
GenerateResponse(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateResponseChoice(
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
@@ -620,10 +621,10 @@ class TestPromptSeededLeadingSpace:
prompt_ids, output_ids, expected = leading_space_ids
wrong_ids = tokenizer.encode("a different prompt", add_special_tokens=False)
choices = await derenderer.derender_chat(
GenerateResponse(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateResponseChoice(
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
@@ -639,10 +640,10 @@ class TestPromptSeededLeadingSpace:
):
_, output_ids, _ = leading_space_ids
choices = await derenderer.derender_chat(
GenerateResponse(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateResponseChoice(
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
@@ -655,10 +656,10 @@ class TestPromptSeededLeadingSpace:
prompt_ids, output_ids, expected = leading_space_ids
choices, _, _ = await derenderer.derender_completion(
[
GenerateResponse(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateResponseChoice(
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
@@ -1494,10 +1495,10 @@ class TestDerenderChatStreamHarmony:
content = "".join(d.content or "" for d in deltas)
batch_choices = await harmony_derenderer.derender_chat(
GenerateResponse(
GenerateTokensResponse(
request_id="test-harmony-batch",
choices=[
GenerateResponseChoice(
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
@@ -1743,10 +1744,10 @@ class TestServingDerenderStreamValidation:
could never fire since derender_chat_stream itself rejects anything
above 1 first."""
serving = self._make_serving(parser_configured=False)
two_choices = GenerateStreamResponse(
two_choices = GenerateTokensStreamResponse(
request_id="t",
choices=[
GenerateResponseStreamChoice(index=i, token_ids=[1]) for i in range(2)
GenerateTokensStreamChoice(index=i, token_ids=[1]) for i in range(2)
],
)
request = DerenderChatStreamRequest(
tests/entrypoints/scale_out/test_factories.py
+33 -2
View File
@@ -6,6 +6,8 @@ from argparse import Namespace
from fastapi import FastAPI
from vllm.entrypoints.launchers.api_server.routers import register_api_routers
from vllm.entrypoints.scale_out.token_in_token_out.api_router import attach_router
from vllm.entrypoints.serve.middleware.authenticate import GUARDED_PREFIX
from vllm.tasks import SupportedTask
RENDER_PATHS = {
@@ -17,7 +19,9 @@ RENDER_PATHS = {
"/v1/responses/render",
}
GENERATE_PATH = "/inference/v1/generate"
SCALE_OUT_PATHS = RENDER_PATHS | {GENERATE_PATH}
ABORT_PATH = "/inference/v1/abort_requests"
UNAUTHENTICATED_ABORT_PATH = "/abort_requests"
SCALE_OUT_PATHS = RENDER_PATHS | {GENERATE_PATH, ABORT_PATH}
def registered_paths(
@@ -66,4 +70,31 @@ def test_render_routes_remain_enabled_for_render_server():
def test_tokens_only_mode_enables_generate_routes_when_flag_is_unset():
paths = registered_paths(("generate",), tokens_only=True)
assert paths >= {GENERATE_PATH, "/abort_requests"}
assert paths >= {GENERATE_PATH, ABORT_PATH, UNAUTHENTICATED_ABORT_PATH}
def test_abort_route_without_tokens_only_requires_api_key():
"""A decode pool that loads a tokenizer serves generate without
--tokens-only and still needs to abort, but only behind --api-key."""
paths = registered_paths(("generate",), enable_scale_out=True, tokens_only=False)
assert ABORT_PATH in paths
assert ABORT_PATH.startswith(GUARDED_PREFIX)
assert UNAUTHENTICATED_ABORT_PATH not in paths
def test_unauthenticated_abort_route_is_not_registered_twice():
"""The RLHF dev router registers /abort_requests before the scale-out
routers, so attaching the scale-out one too would add a shadowed duplicate."""
app = FastAPI()
app.state.args = Namespace(tokens_only=True)
@app.post(UNAUTHENTICATED_ABORT_PATH)
async def existing_abort_requests():
return None
attach_router(app)
paths = [route.path for route in app.routes]
assert paths.count(UNAUTHENTICATED_ABORT_PATH) == 1
assert ABORT_PATH in paths
tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py
+259 -9
View File
@@ -20,7 +20,8 @@ from vllm.entrypoints.scale_out.token_in_token_out.mm_serde import (
)
from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
GenerateRequest,
GenerateResponse,
GenerateTextResponse,
GenerateTokensResponse,
MultiModalFeatures,
PlaceholderRangeInfo,
)
@@ -144,6 +145,7 @@ def _make_request_output(
num_cached_tokens: int | None = None,
index: int = 0,
spec_decode_metrics: RequestSpecDecodeMetrics | None = None,
text: str = "",
) -> RequestOutput:
return RequestOutput(
request_id=request_id,
@@ -153,7 +155,7 @@ def _make_request_output(
outputs=[
CompletionOutput(
index=index,
text="",
text=text,
token_ids=token_ids,
cumulative_logprob=None,
logprobs=logprobs,
@@ -239,7 +241,7 @@ async def test_serve_tokens_skips_mm_cache_for_remote_engine_execution():
response = await serving.serve_tokens(request)
assert isinstance(response, GenerateResponse)
assert isinstance(response, GenerateTokensResponse)
assert (
serving.online_renderer.preprocess_completion.call_args.kwargs["skip_mm_cache"]
is True
@@ -411,7 +413,7 @@ async def test_serve_tokens_returns_spec_decode_metrics(stream: bool):
if isinstance(chunk, dict) and chunk.get("metrics")
)["metrics"]["speculative_decoding"]
else:
assert isinstance(response, GenerateResponse)
assert isinstance(response, GenerateTokensResponse)
assert response.metrics is not None
assert response.metrics.speculative_decoding is not None
payload = response.metrics.speculative_decoding.model_dump()
@@ -511,7 +513,7 @@ async def test_serve_tokens_omits_spec_decode_metrics_for_parallel_sampling(
if isinstance(chunk, dict)
)
else:
assert isinstance(response, GenerateResponse)
assert isinstance(response, GenerateTokensResponse)
assert response.metrics is None
@@ -540,7 +542,7 @@ async def test_non_stream_returns_final_prompt_token_ids_when_requested():
response = await serving.serve_tokens(request)
assert isinstance(response, GenerateResponse)
assert isinstance(response, GenerateTokensResponse)
assert response.prompt_token_ids == final_prompt_token_ids
@@ -622,7 +624,7 @@ async def test_non_stream_prompt_metadata_requires_return_token_ids(
response = await serving.serve_tokens(request)
assert isinstance(response, GenerateResponse)
assert isinstance(response, GenerateTokensResponse)
if return_token_ids:
assert response.prompt_token_ids == [1, 2, 3]
assert response.model_dump()["mm_placeholders"] == MM_PLACEHOLDERS
@@ -652,7 +654,7 @@ async def test_non_stream_text_only_prompt_has_no_mm_placeholders():
response = await serving.serve_tokens(request)
assert isinstance(response, GenerateResponse)
assert isinstance(response, GenerateTokensResponse)
assert response.prompt_token_ids == [1, 2, 3]
assert response.mm_placeholders is None
@@ -678,7 +680,7 @@ async def test_non_stream_encoder_decoder_omits_mm_placeholders():
response = await serving.serve_tokens(request)
assert isinstance(response, GenerateResponse)
assert isinstance(response, GenerateTokensResponse)
assert response.prompt_token_ids == [1, 2, 3]
assert response.mm_placeholders is None
@@ -1195,3 +1197,251 @@ async def test_stream_sampling_mask_matches_each_token_chunk(with_mask):
[[token, token + 1] for token in choice["token_ids"]] if with_mask else None
)
assert choice["sampling_mask"] == expected
async def _collect_sse(response) -> list[Any]:
return _parse_sse_chunks([chunk async for chunk in response])
def _text_request(**kwargs) -> GenerateRequest:
kwargs.setdefault("sampling_params", SamplingParams(max_tokens=10))
return GenerateRequest(
token_ids=[1, 2, 3], model=MODEL_NAME, output_mode="text", **kwargs
)
@pytest.mark.asyncio
async def test_non_stream_tokens_mode_omits_text_and_echoes_output_mode():
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output(
"req-1", token_ids=[10], text="ignored", finish_reason="stop", finished=True
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine)
response = await serving.serve_tokens(
GenerateRequest(
token_ids=[1, 2, 3],
sampling_params=SamplingParams(max_tokens=10),
model=MODEL_NAME,
)
)
assert isinstance(response, GenerateTokensResponse)
dumped = response.model_dump()
assert dumped["output_mode"] == "tokens"
assert "text" not in dumped["choices"][0]
@pytest.mark.asyncio
async def test_non_stream_text_mode_returns_text_and_token_ids():
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output(
"req-1",
token_ids=[10, 20],
text="Hello",
finish_reason="stop",
finished=True,
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine)
response = await serving.serve_tokens(_text_request())
assert isinstance(response, GenerateTextResponse)
assert response.output_mode == "text"
assert response.choices[0].text == "Hello"
assert response.choices[0].token_ids == [10, 20]
assert response.choices[0].finish_reason == "stop"
assert response.usage is not None
assert response.usage.completion_tokens == 2
@pytest.mark.asyncio
@pytest.mark.parametrize("no_tokenizer", ["tokens_only", "skip_tokenizer_init"])
async def test_text_mode_rejected_without_tokenizer(no_tokenizer: str):
"""The no-op detokenizer returns "", so text mode must fail instead
of answering 200 with empty text."""
engine = _mock_engine()
kwargs: dict[str, Any] = {}
if no_tokenizer == "tokens_only":
kwargs["force_no_detokenize"] = True
else:
engine.model_config.skip_tokenizer_init = True
engine.renderer = _build_renderer(engine.model_config)
engine.generate = MagicMock()
serving = _build_serving_tokens(engine, **kwargs)
response = await serving.serve_tokens(_text_request())
assert isinstance(response, ErrorResponse)
assert response.error.code == 400
assert "requires a tokenizer" in response.error.message
engine.generate.assert_not_called()
@pytest.mark.asyncio
async def test_text_mode_logprobs_have_decoded_tokens_and_bytes():
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output(
"req-1",
token_ids=[10],
text="é",
logprobs=[
{
10: Logprob(logprob=-0.5, decoded_token="é"),
11: Logprob(logprob=-1.5, decoded_token="e"),
}
],
finish_reason="stop",
finished=True,
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine)
response = await serving.serve_tokens(
_text_request(sampling_params=SamplingParams(max_tokens=10, logprobs=2))
)
assert isinstance(response, GenerateTextResponse)
logprobs = response.choices[0].logprobs
assert logprobs is not None and logprobs.content is not None
entry = logprobs.content[0]
assert (entry.token, entry.bytes) == ("é", [195, 169])
assert [(t.token, t.bytes) for t in entry.top_logprobs] == [
("é", [195, 169]),
("e", [101]),
]
@pytest.mark.asyncio
async def test_text_mode_logprobs_keep_placeholders_with_return_tokens_as_ids():
"""--return-tokens-as-token-ids keeps token_id:N placeholders, as it does
on /v1/completions, while text is still decoded."""
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output(
"req-1",
token_ids=[10],
text="é",
logprobs=[{10: Logprob(logprob=-0.5, decoded_token="é")}],
finish_reason="stop",
finished=True,
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine, return_tokens_as_token_ids=True)
response = await serving.serve_tokens(
_text_request(sampling_params=SamplingParams(max_tokens=10, logprobs=1))
)
assert isinstance(response, GenerateTextResponse)
assert response.choices[0].text == "é"
logprobs = response.choices[0].logprobs
assert logprobs is not None and logprobs.content is not None
entry = logprobs.content[0]
assert (entry.token, entry.bytes) == ("token_id:10", None)
assert [t.token for t in entry.top_logprobs] == ["token_id:10"]
@pytest.mark.asyncio
async def test_stream_text_mode_emits_text_deltas():
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output("req-1", token_ids=[10], text="Hel")
yield _make_request_output("req-1", token_ids=[20], text="lo")
yield _make_request_output(
"req-1", token_ids=[30], text=" world", finish_reason="stop", finished=True
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine)
response = await serving.serve_tokens(_text_request(stream=True))
parsed = await _collect_sse(response)
assert parsed[-1] == "[DONE]"
data_chunks = parsed[:-1]
assert {c["output_mode"] for c in data_chunks} == {"text"}
assert [c["choices"][0]["text"] for c in data_chunks] == ["Hel", "lo", " world"]
assert [c["choices"][0]["token_ids"] for c in data_chunks] == [[10], [20], [30]]
assert data_chunks[-1]["choices"][0]["finish_reason"] == "stop"
@pytest.mark.asyncio
async def test_stream_text_mode_emits_chunks_without_new_token_ids():
"""Text flushed from the stop string buffer and the final abort output
carry no token IDs but must still reach the client."""
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output("req-1", token_ids=[10], text="a")
yield _make_request_output("req-1", token_ids=[], text="b")
yield _make_request_output(
"req-1", token_ids=[], text="", finish_reason="abort", finished=True
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine)
response = await serving.serve_tokens(_text_request(stream=True))
data_chunks = (await _collect_sse(response))[:-1]
assert [c["choices"][0]["text"] for c in data_chunks] == ["a", "b", ""]
assert data_chunks[1]["choices"][0]["token_ids"] == []
assert data_chunks[2]["choices"][0]["finish_reason"] == "abort"
@pytest.mark.asyncio
async def test_stream_text_mode_skips_outputs_without_text_or_finish():
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output("req-1", token_ids=[10], text="a")
yield _make_request_output("req-1", token_ids=[], text="")
yield _make_request_output(
"req-1", token_ids=[20], text="b", finish_reason="stop", finished=True
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine)
response = await serving.serve_tokens(_text_request(stream=True))
data_chunks = (await _collect_sse(response))[:-1]
assert [c["choices"][0]["text"] for c in data_chunks] == ["a", "b"]
@pytest.mark.asyncio
async def test_stream_text_mode_usage_chunk_echoes_output_mode():
engine = _mock_engine()
async def mock_generate(*args, **kwargs):
yield _make_request_output(
"req-1", token_ids=[10], text="a", finish_reason="stop", finished=True
)
engine.generate = MagicMock(side_effect=mock_generate)
serving = _build_serving_tokens(engine)
response = await serving.serve_tokens(
_text_request(stream=True, stream_options=StreamOptions(include_usage=True))
)
parsed = await _collect_sse(response)
usage_chunk = parsed[-2]
assert usage_chunk["output_mode"] == "text"
assert usage_chunk["choices"] == []
assert usage_chunk["usage"]["completion_tokens"] == 1
tests/entrypoints/scale_out/token_in_token_out/test_protocol.py
+129
View File
@@ -10,9 +10,21 @@ fail loudly if the validator semantics ever drift.
import json
import pytest
from pydantic import TypeAdapter, ValidationError
from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
DerenderChatRequest,
DerenderChatStreamRequest,
DerenderCompletionRequest,
DerenderCompletionStreamRequest,
GenerateRequest,
GenerateResponse,
GenerateStreamResponse,
GenerateTextResponse,
GenerateTextStreamResponse,
GenerateTokensChoice,
GenerateTokensResponse,
GenerateTokensStreamResponse,
MultiModalFeatures,
PlaceholderRangeInfo,
)
@@ -110,3 +122,120 @@ def test_generate_request_rejects_placeholder_outside_prompt():
kwargs_data={"image": ["encoded"]},
),
)
def test_output_mode_defaults_to_tokens():
assert GenerateRequest.model_validate(_base_payload()).output_mode == "tokens"
def test_output_mode_text_is_accepted():
payload = {**_base_payload(), "output_mode": "text"}
assert GenerateRequest.model_validate(payload).output_mode == "text"
@pytest.mark.parametrize("output_mode", ["derender", "bogus", None])
def test_output_mode_rejects_unsupported_levels(output_mode):
"""Levels the server does not implement fail instead of falling back to
tokens which would answer 200 with the wrong response shape."""
payload = {**_base_payload(), "output_mode": output_mode}
with pytest.raises(ValidationError, match="output_mode"):
GenerateRequest.model_validate(payload)
def test_text_mode_rejects_detokenize_false():
payload = {
"token_ids": [1, 2, 3],
"sampling_params": {"detokenize": False},
"output_mode": "text",
}
with pytest.raises(ValidationError, match="detokenize"):
GenerateRequest.model_validate(payload)
def test_tokens_mode_allows_detokenize_false():
payload = {
"token_ids": [1, 2, 3],
"sampling_params": {"detokenize": False},
"output_mode": "tokens",
}
assert GenerateRequest.model_validate(payload).output_mode == "tokens"
def test_response_without_output_mode_parses_as_tokens():
"""Existing clients and older servers never send output_mode."""
parsed = TypeAdapter(GenerateResponse).validate_python(
{"choices": [{"index": 0, "token_ids": [1]}]}
)
assert isinstance(parsed, GenerateTokensResponse)
assert parsed.output_mode == "tokens"
def test_stream_chunk_without_output_mode_parses_as_tokens():
parsed = TypeAdapter(GenerateStreamResponse).validate_python(
{"choices": [{"index": 0, "token_ids": [1]}]}
)
assert isinstance(parsed, GenerateTokensStreamResponse)
def test_text_response_and_chunk_dispatch_on_output_mode():
response = TypeAdapter(GenerateResponse).validate_python(
{"output_mode": "text", "choices": [{"index": 0, "text": "hi"}]}
)
chunk = TypeAdapter(GenerateStreamResponse).validate_python(
{"output_mode": "text", "choices": [{"index": 0, "text": "hi"}]}
)
assert isinstance(response, GenerateTextResponse)
assert isinstance(chunk, GenerateTextStreamResponse)
def test_text_response_requires_text_on_every_choice():
with pytest.raises(ValidationError, match="text"):
TypeAdapter(GenerateResponse).validate_python(
{"output_mode": "text", "choices": [{"index": 0}]}
)
@pytest.mark.parametrize("output_mode", ["derender", None, 5])
def test_response_rejects_unknown_output_mode(output_mode):
with pytest.raises(ValidationError):
TypeAdapter(GenerateResponse).validate_python(
{"output_mode": output_mode, "choices": []}
)
def test_derender_requests_accept_responses_without_output_mode():
DerenderChatRequest.model_validate(
{"generate_response": {"choices": [{"index": 0, "token_ids": [1]}]}}
)
DerenderChatStreamRequest.model_validate(
{
"stream": True,
"generate_chunk": {"choices": [{"index": 0, "token_ids": [1]}]},
}
)
def test_derender_requests_reject_text_responses():
"""Text responses are already detokenized with stop strings applied, so
re-decoding their token_ids would silently undo the truncation."""
text = {"output_mode": "text", "choices": [{"index": 0, "text": "hi"}]}
with pytest.raises(ValidationError, match="output_mode"):
DerenderChatRequest.model_validate({"generate_response": text})
with pytest.raises(ValidationError, match="output_mode"):
DerenderCompletionRequest.model_validate({"generate_responses": [text]})
with pytest.raises(ValidationError, match="output_mode"):
DerenderChatStreamRequest.model_validate(
{"stream": True, "generate_chunk": text}
)
with pytest.raises(ValidationError, match="output_mode"):
DerenderCompletionStreamRequest.model_validate(
{"stream": True, "generate_chunk": text}
)
def test_tokens_response_echoes_output_mode_without_text():
dumped = GenerateTokensResponse(
choices=[GenerateTokensChoice(index=0, token_ids=[1])]
).model_dump()
assert dumped["output_mode"] == "tokens"
assert "text" not in dumped["choices"][0]
tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py
+185
View File
@@ -19,6 +19,7 @@ from vllm.v1.engine.detokenizer import check_stop_strings
MODEL_NAME = "Qwen/Qwen3-0.6B"
GEN_ENDPOINT = "/inference/v1/generate"
ABORT_ENDPOINT = "/inference/v1/abort_requests"
def get_vocab_size(model_name):
@@ -590,3 +591,187 @@ async def test_generate_with_lora_adapter(client, tokenizer, messages):
completions_res = completions_data["choices"][0]["message"]["content"]
assert generate_res == completions_res
def _chat_prompt_token_ids(tokenizer, messages) -> list[int]:
return tokenizer.apply_chat_template(
messages,
add_generation_prompt=True,
enable_thinking=False, # default with Qwen3
return_dict=True, # default with Transformers v5
).input_ids
def _text_mode_payload(token_ids, **sampling_params) -> dict:
return {
"model": MODEL_NAME,
"token_ids": token_ids,
"sampling_params": {"max_tokens": 24, "temperature": 0.0, **sampling_params},
"output_mode": "text",
"stream": False,
}
@pytest.mark.asyncio
@pytest.mark.skipif(
envs.VLLM_USE_RUST_FRONTEND,
reason="output_mode is not supported by the Rust frontend",
)
async def test_text_mode_matches_chat_completions(client, tokenizer, messages):
token_ids = _chat_prompt_token_ids(tokenizer, messages)
resp = await client.post(GEN_ENDPOINT, json=_text_mode_payload(token_ids))
resp.raise_for_status()
data = resp.json()
assert data["output_mode"] == "text"
payload = {
"model": MODEL_NAME,
"messages": messages,
"max_tokens": 24,
"temperature": 0.0,
"stream": False,
"chat_template_kwargs": {"enable_thinking": False},
}
chat_resp = await client.post("/v1/chat/completions", json=payload)
chat_content = chat_resp.json()["choices"][0]["message"]["content"]
assert data["choices"][0]["text"] == chat_content
assert data["choices"][0]["token_ids"]
@pytest.mark.asyncio
@pytest.mark.skipif(
envs.VLLM_USE_RUST_FRONTEND,
reason="output_mode is not supported by the Rust frontend",
)
async def test_text_mode_stream_matches_non_stream(client, tokenizer, messages):
token_ids = _chat_prompt_token_ids(tokenizer, messages)
payload = _text_mode_payload(token_ids)
non_stream = await client.post(GEN_ENDPOINT, json=payload)
non_stream.raise_for_status()
chunks = []
async with client.stream(
"POST", GEN_ENDPOINT, json={**payload, "stream": True}
) as resp:
resp.raise_for_status()
async for line in resp.aiter_lines():
if line.startswith("data: ") and line != "data: [DONE]":
chunks.append(json.loads(line[len("data: ") :]))
assert {chunk["output_mode"] for chunk in chunks} == {"text"}
assert chunks[-1]["choices"][0]["finish_reason"] is not None
streamed_text = "".join(chunk["choices"][0]["text"] for chunk in chunks)
assert streamed_text == non_stream.json()["choices"][0]["text"]
@pytest.mark.asyncio
@pytest.mark.skipif(
envs.VLLM_USE_RUST_FRONTEND,
reason="output_mode is not supported by the Rust frontend",
)
@pytest.mark.parametrize(
"overrides, expected",
[
pytest.param(
{"sampling_params": {"detokenize": False}}, "detokenize", id="no-detok"
),
pytest.param({"output_mode": "derender"}, "output_mode", id="unsupported"),
],
)
async def test_text_mode_invalid_requests_return_400(client, overrides, expected):
payload = {
**_text_mode_payload([1, 2, 3], max_tokens=5),
**overrides,
}
resp = await client.post(GEN_ENDPOINT, json=payload)
assert resp.status_code == 400
assert expected in resp.json()["error"]["message"]
@pytest.mark.asyncio
@pytest.mark.skipif(
envs.VLLM_USE_RUST_FRONTEND,
reason="--tokens-only is not supported by the Rust frontend",
)
@pytest.mark.parametrize("server", [["--tokens-only"]], indirect=True)
async def test_text_mode_rejected_when_tokens_only(client):
"""Without a tokenizer the detokenizer returns "", so the request must
fail instead of answering 200 with empty text."""
resp = await client.post(GEN_ENDPOINT, json=_text_mode_payload([1, 2, 3]))
assert resp.status_code == 400
assert "requires a tokenizer" in resp.json()["error"]["message"]
payload = {**_text_mode_payload([1, 2, 3]), "output_mode": "tokens"}
payload["sampling_params"]["detokenize"] = False
resp = await client.post(GEN_ENDPOINT, json=payload)
resp.raise_for_status()
@pytest.mark.asyncio
@pytest.mark.skipif(
envs.VLLM_USE_RUST_FRONTEND,
reason="output_mode is not supported by the Rust frontend",
)
async def test_text_mode_stream_delivers_abort_finish_chunk(
client, tokenizer, messages
):
"""The final output after an abort has no new token IDs, so it
reaches a text stream only because text mode emits finish-only chunks."""
payload = {
**_text_mode_payload(
_chat_prompt_token_ids(tokenizer, messages),
max_tokens=900,
ignore_eos=True,
),
"stream": True,
"request_id": "text-abort-e2e",
}
chunks = []
aborted = False
async with client.stream("POST", GEN_ENDPOINT, json=payload) as resp:
resp.raise_for_status()
async for line in resp.aiter_lines():
if not line.startswith("data: ") or line == "data: [DONE]":
continue
chunk = json.loads(line[len("data: ") :])
chunks.append(chunk)
if not aborted and chunk["choices"] and chunk["choices"][0]["text"]:
abort = await client.post(
ABORT_ENDPOINT,
json={"request_ids": [chunk["request_id"]]},
)
assert abort.status_code == 200
aborted = True
assert aborted
assert {chunk["output_mode"] for chunk in chunks} == {"text"}
final_choice = chunks[-1]["choices"][0]
assert final_choice["finish_reason"] == "abort"
generated = sum(len(chunk["choices"][0]["token_ids"] or []) for chunk in chunks)
assert generated < 900
@pytest.mark.asyncio
@pytest.mark.skipif(
envs.VLLM_USE_RUST_FRONTEND,
reason="the Rust frontend does not serve /inference/v1/abort_requests",
)
async def test_abort_requests_is_served_without_tokens_only(client):
resp = await client.post(
ABORT_ENDPOINT, json={"request_ids": ["generate-tokens-unknown"]}
)
assert resp.status_code == 200
resp = await client.post(ABORT_ENDPOINT, json={})
assert resp.status_code == 400
# The unauthenticated path is only served with --tokens-only.
resp = await client.post(
"/abort_requests", json={"request_ids": ["generate-tokens-unknown"]}
)
assert resp.status_code == 404
tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py
+49
View File
@@ -1,6 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from unittest.mock import MagicMock
from vllm.entrypoints.scale_out.token_in_token_out.serving import ServingTokens
from vllm.logprobs import Logprob
@@ -36,3 +38,50 @@ def test_logprobs_minus_one_emits_all_tokens():
num_output_top_logprobs=-1,
)
assert len(result.content[0].top_logprobs) == 2
def test_tokens_level_logprobs_have_no_bytes():
result = ServingTokens._create_tokens_logprobs(
None,
token_ids=[7],
top_logprobs=[{7: Logprob(-0.9), 8: Logprob(-1.1)}],
num_output_top_logprobs=2,
)
entry = result.content[0]
assert entry.bytes is None
assert all(t.bytes is None for t in entry.top_logprobs)
def test_resolved_logprobs_use_engine_decoded_tokens_and_bytes():
tokenizer = MagicMock()
result = ServingTokens._create_tokens_logprobs(
None,
token_ids=[7],
top_logprobs=[
{7: Logprob(-0.9, decoded_token="é"), 8: Logprob(-1.1, decoded_token="e")}
],
num_output_top_logprobs=2,
tokenizer=tokenizer,
)
entry = result.content[0]
assert (entry.token, entry.bytes) == ("é", [195, 169])
assert [(t.token, t.bytes) for t in entry.top_logprobs] == [
("é", [195, 169]),
("e", [101]),
]
tokenizer.decode.assert_not_called()
def test_resolved_logprobs_decode_tokens_the_engine_did_not():
tokenizer = MagicMock()
tokenizer.decode.side_effect = lambda ids: f"<{ids[0]}>"
result = ServingTokens._create_tokens_logprobs(
None,
token_ids=[7, 9],
top_logprobs=[{7: Logprob(-0.9)}, None],
num_output_top_logprobs=1,
tokenizer=tokenizer,
)
assert [e.token for e in result.content] == ["<7>", "<9>"]
assert result.content[0].top_logprobs[0].token == "<7>"
assert result.content[1].bytes == list(b"<9>")
vllm/entrypoints/launchers/cli_args.py
+3 -3
View File
@@ -207,9 +207,9 @@ class BaseFrontendArgs:
enable_scale_out: bool = False
"""
If set to True, register the scale-out endpoints (`/render`, `/derender`,
and `/inference/v1/generate`) on `vllm serve`. Has no effect on
`vllm launch render` or `vllm serve --tokens-only`, which always register
their required endpoints regardless of this flag.
`/inference/v1/generate` and `/inference/v1/abort_requests`) on `vllm serve`.
Has no effect on `vllm launch render` or `vllm serve --tokens-only` which
always register their required endpoints regardless of this flag.
"""
fingerprint_mode: Literal["full", "hash", "custom", "none"] = "full"
"""Controls the ``system_fingerprint`` field on responses.
vllm/entrypoints/scale_out/derender/serving.py
+2 -2
View File
@@ -29,7 +29,7 @@ from ..token_in_token_out.protocol import (
DerenderCompletionRequest,
DerenderCompletionStreamRequest,
DerenderStreamState,
GenerateResponse,
GenerateTokensResponse,
MultiModalFeatures,
)
@@ -54,7 +54,7 @@ class ServingDerender(BaseServing):
def _validate_derender_bounds(
self,
generate_responses: list[GenerateResponse],
generate_responses: list[GenerateTokensResponse],
) -> ErrorResponse | None:
"""Reject derender payloads that exceed resource bounds.
vllm/entrypoints/scale_out/token_in_token_out/api_router.py
+42 -26
View File
@@ -21,7 +21,7 @@ from vllm.logger import init_logger
from ...serve.engine.protocol import ErrorResponse
from .protocol import (
GenerateRequest,
GenerateResponse,
GenerateResponseBase,
)
from .serving import ServingTokens
@@ -67,35 +67,51 @@ async def generate(request: GenerateRequest, raw_request: Request):
content=generator.model_dump(), status_code=generator.error.code
)
elif isinstance(generator, GenerateResponse):
elif isinstance(generator, GenerateResponseBase):
return JSONResponse(content=generator.model_dump())
return StreamingResponse(content=generator, media_type="text/event-stream")
abort_router = APIRouter()
async def abort_requests(raw_request: Request):
"""Abort one or more requests. To be used in a
Disaggregated Everything setup.
"""
try:
body = await raw_request.json()
except json.JSONDecodeError as e:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST.value,
detail=f"JSON decode error: {e}",
) from e
request_ids = body.get("request_ids")
if request_ids is None:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST.value,
detail="Missing 'request_ids' in request body",
)
# Abort requests in background
asyncio.create_task(engine_client(raw_request).abort(request_ids))
return Response(status_code=200)
# Under the `/inference` prefix, so `--api-key` guards it.
router.add_api_route("/inference/v1/abort_requests", abort_requests, methods=["POST"])
# Unauthenticated legacy path, kept for existing `--tokens-only` deployments.
abort_router.add_api_route("/abort_requests", abort_requests, methods=["POST"])
def attach_router(app: FastAPI):
if getattr(app.state.args, "tokens_only", False):
@router.post("/abort_requests")
async def abort_requests(raw_request: Request):
"""Abort one or more requests. To be used in a
Disaggregated Everything setup.
"""
try:
body = await raw_request.json()
except json.JSONDecodeError as e:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST.value,
detail=f"JSON decode error: {e}",
) from e
request_ids = body.get("request_ids")
if request_ids is None:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST.value,
detail="Missing 'request_ids' in request body",
)
# Abort requests in background
asyncio.create_task(engine_client(raw_request).abort(request_ids))
return Response(status_code=200)
app.include_router(router)
if not getattr(app.state.args, "tokens_only", False):
return
# The RLHF dev router registers its own /abort_requests first. Registering
# this one too would only add a shadowed route with a duplicate operation ID.
has_abort_route = any(
getattr(route, "path", None) == "/abort_requests" for route in app.routes
)
if not has_abort_route:
app.include_router(abort_router)
vllm/entrypoints/scale_out/token_in_token_out/protocol.py
+149 -48
View File
@@ -4,9 +4,11 @@ from typing import Annotated, Any, Literal, TypeAlias
from pydantic import (
BaseModel,
Discriminator,
Field,
NonNegativeInt,
PrivateAttr,
Tag,
field_validator,
model_validator,
)
@@ -34,6 +36,8 @@ from vllm.utils import random_uuid
####### Tokens IN <> Tokens OUT #######
OutputMode: TypeAlias = Literal["tokens", "text"]
class PlaceholderRangeInfo(BaseModel):
"""Serializable placeholder location for a single multi-modal item."""
@@ -44,7 +48,7 @@ class PlaceholderRangeInfo(BaseModel):
length: int = Field(gt=0)
"""Number of placeholder tokens."""
# TODO: add ``is_embed: list[bool] | None`` once the /generate side
# TODO: add `is_embed: list[bool] | None` once the /generate side
# consumes features — some models (e.g. Qwen-VL) use sparse
# placeholder masks that cannot be recomputed from offset+length alone.
@@ -74,22 +78,22 @@ class MultiModalFeatures(BaseModel):
kwargs_data: dict[str, list[str | None]] | None = None
"""Per-modality serialized tensor data.
Each value is a list parallel to ``mm_hashes[modality]``. A ``str``
entry is a base64-encoded ``MultiModalKwargsItem``; ``None`` means
Each value is a list parallel to `mm_hashes[modality]`. A `str`
entry is a base64-encoded `MultiModalKwargsItem`; `None` means
the item should be resolved from cache. The entire field is
``None`` for metadata-only (cache-hit) responses.
`None` for metadata-only (cache-hit) responses.
"""
mm_metadata: dict[str, list[str | None]] | None = None
"""Per-modality serialized metadata for disaggregated prefill.
Each value is a list parallel to ``mm_hashes[modality]``. A ``str``
entry is a base64-encoded ``MultiModalKwargsItem`` containing only
placeholder-metadata and ``keep_on_cpu`` fields. ``None`` means that
Each value is a list parallel to `mm_hashes[modality]`. A `str`
entry is a base64-encoded `MultiModalKwargsItem` containing only
placeholder-metadata and `keep_on_cpu` fields. `None` means that
the metadata is unavailable for that item. Prefill can use this
instead of ``kwargs_data`` only when ``ec_transfer_params`` is also
instead of `kwargs_data` only when `ec_transfer_params` is also
set, so embeddings arrive through the EC connector rather than from
``pixel_values``.
`pixel_values`.
"""
@model_validator(mode="after")
@@ -174,7 +178,7 @@ class GenerateRequest(BaseModel):
content_parts: list[dict[str, Any]] | None = None
"""Raw multimodal input; server resolves media. Mutually exclusive
with ``features``."""
with `features`."""
@model_validator(mode="after")
def _check_mm_fields_exclusive(self) -> "GenerateRequest":
@@ -217,6 +221,16 @@ class GenerateRequest(BaseModel):
),
)
output_mode: OutputMode = Field(
default="tokens",
description=(
"Response level. 'tokens' returns token IDs only. 'text' also "
"returns the detokenized text and logprobs with decoded token "
"strings. 'text' needs a server that loads a tokenizer and "
"sampling_params.detokenize to be true."
),
)
stream: bool | None = False
stream_options: StreamOptions | None = None
cache_salt: str | None = Field(
@@ -253,12 +267,12 @@ class GenerateRequest(BaseModel):
),
)
# Tracks which keys the caller explicitly set inside ``sampling_params``
# Tracks which keys the caller explicitly set inside `sampling_params`
# when the request was parsed from a JSON body. Lets the server tell
# "client said max_tokens=16" from "client said nothing → dataclass
# default 16" so it can apply server-side defaulting only in the latter
# case. ``None`` means the request was constructed with a pre-built
# ``SamplingParams`` instance (e.g. from internal callers that have
# case. `None` means the request was constructed with a pre-built
# `SamplingParams` instance (e.g. from internal callers that have
# already resolved values), in which case all fields are considered set.
_sampling_params_provided_keys: set[str] | None = PrivateAttr(default=None)
_response_mm_placeholders: dict[str, list[PlaceholderRangeInfo]] | None = (
@@ -284,6 +298,15 @@ class GenerateRequest(BaseModel):
validate_cache_salt(data.get("cache_salt"))
return data
@model_validator(mode="after")
def _check_output_mode_detokenizes(self) -> "GenerateRequest":
if self.output_mode != "tokens" and not self.sampling_params.detokenize:
raise ValueError(
f"output_mode={self.output_mode!r} requires "
"sampling_params.detokenize to be true"
)
return self
@model_validator(mode="after")
def _validate_multimodal_feature_bounds(self) -> "GenerateRequest":
if self.features is None:
@@ -299,10 +322,10 @@ class GenerateRequest(BaseModel):
return self
def is_sampling_param_provided(self, name: str) -> bool:
"""Whether the caller explicitly set ``sampling_params.<name>``.
"""Whether the caller explicitly set `sampling_params.<name>`.
For requests parsed from a JSON body, this reflects the raw input
dict. For requests constructed with a pre-built ``SamplingParams``
dict. For requests constructed with a pre-built `SamplingParams`
instance, all fields are considered provided so server-side defaults
do not clobber values already resolved upstream.
"""
@@ -317,21 +340,22 @@ class GenerateRequest(BaseModel):
)
class GenerateResponseChoice(BaseModel):
class GenerateChoiceBase(BaseModel):
"""Fields shared by every `output_mode` of a non-streaming choice."""
index: int
logprobs: ChatCompletionLogProbs | None = None
# per OpenAI spec this is the default
finish_reason: str | None = "stop"
token_ids: list[int] | None = None
# Per-token expert routing decisions, base64-encoded ``.npy`` bytes
# Per-token expert routing decisions, base64-encoded `.npy` bytes
# (numpy serialization). Shape after decode:
# (num_tokens - 1, num_layers, num_experts_per_tok) dtype uint8/uint16/int32
# ``num_tokens - 1`` because the last sampled token has not been
# `num_tokens - 1` because the last sampled token has not been
# forwarded yet and therefore has no routing data.
# Decode:
# np.load(io.BytesIO(base64.b64decode(s)))
# ``None`` if (a) the request was aborted before any forward pass,
# or (b) ``enable_return_routed_experts`` is off server-side.
# `None` if (a) the request was aborted before any forward pass,
# or (b) `enable_return_routed_experts` is off server-side.
routed_experts: str | None = None
sampling_mask: list[list[int]] | None = None
@@ -343,16 +367,43 @@ class GenerateResponseChoice(BaseModel):
return v
class GenerateResponseStreamChoice(BaseModel):
index: int
class GenerateTokensChoice(GenerateChoiceBase):
logprobs: ChatCompletionLogProbs | None = None
"""Logprobs whose tokens are `token_id:N` placeholders."""
class GenerateTextChoice(GenerateChoiceBase):
text: str
"""Detokenized output. Excludes a matched stop string unless
`include_stop_str_in_output` is set, while `token_ids` keeps every
generated token."""
logprobs: ChatCompletionLogProbs | None = None
"""Logprobs with decoded token strings and `bytes`."""
class GenerateStreamChoiceBase(BaseModel):
"""Fields shared by every `output_mode` of a streaming choice."""
index: int
finish_reason: str | None = None
token_ids: list[int] | None = None
routed_experts: str | None = None
sampling_mask: list[list[int]] | None = None
class GenerateStreamResponse(BaseModel):
class GenerateTokensStreamChoice(GenerateStreamChoiceBase):
logprobs: ChatCompletionLogProbs | None = None
class GenerateTextStreamChoice(GenerateStreamChoiceBase):
text: str
"""Text delta since the previous chunk of this choice."""
logprobs: ChatCompletionLogProbs | None = None
class GenerateStreamResponseBase(BaseModel):
request_id: str = Field(
default_factory=lambda: f"{random_uuid()}",
description=(
@@ -361,14 +412,23 @@ class GenerateStreamResponse(BaseModel):
"through out the inference process and return in response."
),
)
choices: list[GenerateResponseStreamChoice]
usage: UsageInfo | None = Field(default=None)
prompt_token_ids: list[NonNegativeInt] | None = None
mm_placeholders: dict[str, list[PlaceholderRangeInfo]] | None = None
metrics: PerRequestMetrics | None = None
class GenerateResponse(BaseModel):
class GenerateTokensStreamResponse(GenerateStreamResponseBase):
output_mode: Literal["tokens"] = "tokens"
choices: list[GenerateTokensStreamChoice]
class GenerateTextStreamResponse(GenerateStreamResponseBase):
output_mode: Literal["text"] = "text"
choices: list[GenerateTextStreamChoice]
class GenerateResponseBase(BaseModel):
request_id: str = Field(
default_factory=lambda: f"{random_uuid()}",
description=(
@@ -379,7 +439,6 @@ class GenerateResponse(BaseModel):
)
model: str | None = None
created: int | None = None
choices: list[GenerateResponseChoice]
usage: UsageInfo | None = Field(default=None)
prompt_logprobs: list[dict[int, Logprob] | None] | None = None
prompt_token_id_logprobs: str | None = None
@@ -399,6 +458,42 @@ class GenerateResponse(BaseModel):
)
class GenerateTokensResponse(GenerateResponseBase):
output_mode: Literal["tokens"] = "tokens"
choices: list[GenerateTokensChoice]
class GenerateTextResponse(GenerateResponseBase):
output_mode: Literal["text"] = "text"
choices: list[GenerateTextChoice]
def output_mode_or_tokens(value: Any) -> Any:
"""Discriminator for `GenerateResponse` and `GenerateStreamResponse`.
A missing `output_mode` means `tokens`. Existing clients and servers
that predate the field never send it.
"""
if isinstance(value, dict):
return value.get("output_mode", "tokens")
return getattr(value, "output_mode", "tokens")
# Type aliases, not classes: use `GenerateResponseBase` or
# `GenerateStreamResponseBase` for `isinstance` checks and the
# `Generate<Level>...` classes to construct.
GenerateResponse: TypeAlias = Annotated[
Annotated[GenerateTokensResponse, Tag("tokens")]
| Annotated[GenerateTextResponse, Tag("text")],
Discriminator(output_mode_or_tokens),
]
GenerateStreamResponse: TypeAlias = Annotated[
Annotated[GenerateTokensStreamResponse, Tag("tokens")]
| Annotated[GenerateTextStreamResponse, Tag("text")],
Discriminator(output_mode_or_tokens),
]
class DerenderChatRequest(BaseModel):
"""Request for the /v1/chat/completions/derender endpoint (non-streaming).
@@ -412,8 +507,13 @@ class DerenderChatRequest(BaseModel):
model: str | None = None
"""Served model name. Defaults to the server's served model name."""
generate_response: GenerateResponse
"""The complete token-in / token-out engine response to derender."""
generate_response: GenerateTokensResponse
"""The complete token-in / token-out engine response to derender.
Only `output_mode="tokens"` responses are accepted. Other modes are
already detokenized and re-decoding their `token_ids` would undo the
engine's stop string truncation.
"""
prompt_tokens: int | None = None
"""Prompt token count for usage; defaults to 0 if omitted.
@@ -455,9 +555,9 @@ class DerenderCompletionRequest(BaseModel):
model: str | None = None
"""Served model name. Defaults to the server's served model name."""
generate_responses: list[GenerateResponse]
"""One response per prompt, parallel to the list[GenerateRequest]
returned by /v1/completions/render."""
generate_responses: list[GenerateTokensResponse]
"""One `output_mode="tokens"` response per prompt, parallel to the
list[GenerateRequest] returned by /v1/completions/render."""
prompt_tokens: list[int] | None = None
"""One prompt token count per response; each defaults to 0 if omitted.
@@ -589,7 +689,7 @@ class DerenderStreamState(BaseModel):
token in here through `parse_delta` (discarding the result) before
processing the current chunk's tokens since parser internal state
cannot be serialized into this stateless model. Unavoidably O(n)
bounded by ``max_model_len`` (enforced server side, not by a field
bounded by `max_model_len` (enforced server side, not by a field
validator here since the bound is model dependent).
"""
@@ -630,24 +730,25 @@ class DerenderChatStreamRequest(BaseModel):
"""One chunk streaming derender request for /v1/chat/completions/derender.
The client sends one request per SSE chunk received from
``/inference/v1/generate``. Each request carries the generate chunk
plus the ``stream_state`` returned by the previous call (``None`` on the
`/inference/v1/generate`. Each request carries the generate chunk
plus the `stream_state` returned by the previous call (`None` on the
first call). The response contains the derendered chunk and the updated
state to be passed to the next call.
This implements stateless no server side session. All mutable state lives in
the client carried ``stream_state``.
the client carried `stream_state`.
"""
# --8<-- [start:derender-chat-stream-request]
stream: Literal[True]
model: str | None = None
generate_chunk: GenerateStreamResponse
"""One SSE chunk from ``/inference/v1/generate`` (``stream=True``)."""
generate_chunk: GenerateTokensStreamResponse
"""One `output_mode="tokens"` SSE chunk from `/inference/v1/generate`
(`stream=True`)."""
stream_state: DerenderStreamState | None = None
"""Client carried detok state from the previous call. ``None`` on first."""
"""Client carried detok state from the previous call. `None` on first."""
prompt_tokens: int | None = None
"""Prompt token count for usage. Forwarded from the render step."""
@@ -655,7 +756,7 @@ class DerenderChatStreamRequest(BaseModel):
prompt_token_ids: list[NonNegativeInt] | None = None
"""Prompt token IDs. Required by the parser path's `parse_delta` to
settle its initial reasoning state (e.g. chat templates that pre-open
``<think>``). `prompt_tokens` is a usage count and cannot serve this
`<think>`). `prompt_tokens` is a usage count and cannot serve this
purpose. Sourced from `GenerateRequest.token_ids` at the render step.
Rejected with a 400 (by `ServingDerender`) when a tool or reasoning
@@ -680,7 +781,7 @@ class DerenderChatStreamRequest(BaseModel):
class DerenderCompletionStreamRequest(BaseModel):
"""One chunk streaming derender request for /v1/completions/derender.
Parallel to ``DerenderChatStreamRequest`` for the completions endpoint.
Parallel to `DerenderChatStreamRequest` for the completions endpoint.
Each call processes one SSE chunk (one output sequence's delta) and
returns the derendered chunk plus updated state.
"""
@@ -689,11 +790,11 @@ class DerenderCompletionStreamRequest(BaseModel):
stream: Literal[True]
model: str | None = None
generate_chunk: GenerateStreamResponse
"""One SSE chunk from ``/inference/v1/generate``."""
generate_chunk: GenerateTokensStreamResponse
"""One `output_mode="tokens"` SSE chunk from `/inference/v1/generate`."""
stream_state: DerenderStreamState | None = None
"""Client-carried detok state. ``None`` on the first call."""
"""Client-carried detok state. `None` on the first call."""
prompt_tokens: int | None = None
"""Prompt token count for usage."""
@@ -723,16 +824,16 @@ class DerenderChatStreamResponse(BaseModel):
class DerenderCompletionStreamResponse(BaseModel):
"""Response for one streaming completions derender chunk.
Parallel to ``DerenderChatStreamResponse`` for the completions endpoint.
Parallel to `DerenderChatStreamResponse` for the completions endpoint.
"""
chunk: CompletionStreamResponse
stream_state: DerenderStreamState
# Determines the type by checking the ``stream`` field's literal value. A body without
# ``stream`` validates as the non-streaming member
# (``stream`` defaults to ``False`` there), so FastAPI can validate and dispatch both
# Determines the type by checking the `stream` field's literal value. A body without
# `stream` validates as the non-streaming member
# (`stream` defaults to `False` there), so FastAPI can validate and dispatch both
# shapes on a single path.
DerenderChatRequestUnion: TypeAlias = DerenderChatRequest | DerenderChatStreamRequest
DerenderCompletionRequestUnion: TypeAlias = (
vllm/entrypoints/scale_out/token_in_token_out/serving.py
+146 -44
View File
@@ -6,6 +6,7 @@ import asyncio
import time
from collections.abc import AsyncGenerator
from collections.abc import Sequence as GenericSequence
from typing import Any
import msgspec
from fastapi import Request
@@ -20,6 +21,7 @@ from vllm.entrypoints.generate.base.serving import (
GenerateBaseServing,
build_spec_decoding_metrics,
clamp_prompt_logprobs,
format_token_id_placeholder,
)
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionLogProb,
@@ -46,6 +48,7 @@ from vllm.multimodal.inputs import (
from vllm.outputs import RequestOutput
from vllm.renderers.online_renderer import OnlineRenderer
from vllm.sampling_params import RequestOutputKind, SamplingParams
from vllm.tokenizers import TokenizerLike
from vllm.utils.collection_utils import as_list
from vllm.utils.serial_utils import numpy2base64
@@ -56,14 +59,45 @@ from .mm_features import (
from .protocol import (
GenerateRequest,
GenerateResponse,
GenerateResponseChoice,
GenerateResponseStreamChoice,
GenerateStreamResponse,
GenerateTextChoice,
GenerateTextResponse,
GenerateTextStreamChoice,
GenerateTextStreamResponse,
GenerateTokensChoice,
GenerateTokensResponse,
GenerateTokensStreamChoice,
GenerateTokensStreamResponse,
)
logger = init_logger(__name__)
def _logprob_token(
token_id: int, logprob: Logprob | None, tokenizer: TokenizerLike | None
) -> tuple[str, list[int] | None]:
"""Token string and UTF-8 bytes for one logprob entry.
Without a tokenizer the token is a ``token_id:N`` placeholder with no bytes.
"""
if tokenizer is None:
return format_token_id_placeholder(token_id), None
token = (
tokenizer.decode([token_id])
if logprob is None
else GenerateBaseServing._get_decoded_token(logprob, token_id, tokenizer)
)
return token, list(token.encode("utf-8", errors="replace"))
def _top_logprob(
token_id: int, logprob: Logprob, tokenizer: TokenizerLike | None
) -> ChatCompletionLogProb:
token, token_bytes = _logprob_token(token_id, logprob, tokenizer)
return ChatCompletionLogProb(
token=token, logprob=max(logprob.logprob, -9999.0), bytes=token_bytes
)
class ServingTokens(GenerateBaseServing):
"""Provides Tokens IN <> Tokens OUT functionality to vLLM API."""
@@ -89,6 +123,9 @@ class ServingTokens(GenerateBaseServing):
self.enable_prompt_tokens_details = enable_prompt_tokens_details
self.enable_log_outputs = enable_log_outputs
self.force_no_detokenize = force_no_detokenize
self.has_tokenizer = (
not force_no_detokenize and self.renderer.tokenizer is not None
)
if force_no_detokenize:
logger.info(
"Tokens-only mode is enabled, skipping detokenization "
@@ -171,6 +208,15 @@ class ServingTokens(GenerateBaseServing):
"because detokenization is disabled. Check stop strings on "
"the coordinator, or use stop_token_ids."
)
if request.output_mode != "tokens" and not self.has_tokenizer:
# The no-op detokenizer returns "", so without this check the
# request would succeed with empty text.
return self.create_error_response(
f"output_mode={request.output_mode!r} requires a tokenizer, but "
"this server does not load one (--tokens-only or "
"--skip-tokenizer-init). Send the request to a server that "
"loads a tokenizer, or use output_mode='tokens'."
)
try:
msgspec.msgpack.encode(
(
@@ -331,6 +377,8 @@ class ServingTokens(GenerateBaseServing):
created_time = int(time.time())
final_res: RequestOutput | None = None
sampling_params: SamplingParams = request.sampling_params
text_mode = request.output_mode == "text"
tokenizer = self._logprobs_tokenizer(text_mode)
try:
async for res in result_generator:
@@ -340,7 +388,8 @@ class ServingTokens(GenerateBaseServing):
assert final_res is not None
choices: list[GenerateResponseChoice] = []
tokens_choices: list[GenerateTokensChoice] = []
text_choices: list[GenerateTextChoice] = []
num_generated_tokens = 0
for output in final_res.outputs:
self._raise_if_error(output.finish_reason, request_id)
@@ -355,6 +404,7 @@ class ServingTokens(GenerateBaseServing):
token_ids=token_ids,
top_logprobs=out_logprobs,
num_output_top_logprobs=sampling_params.logprobs,
tokenizer=tokenizer,
)
else:
logprobs = None
@@ -369,7 +419,7 @@ class ServingTokens(GenerateBaseServing):
if output.sampling_mask is not None:
sampling_mask = output.sampling_mask.token_ids
choice_data = GenerateResponseChoice(
choice_fields: dict[str, Any] = dict(
index=output.index,
logprobs=logprobs,
finish_reason=output.finish_reason if output.finish_reason else "stop",
@@ -377,8 +427,12 @@ class ServingTokens(GenerateBaseServing):
routed_experts=routed_experts_b64,
sampling_mask=sampling_mask,
)
choices.append(choice_data)
if text_mode:
text_choices.append(
GenerateTextChoice(text=output.text, **choice_fields)
)
else:
tokens_choices.append(GenerateTokensChoice(**choice_fields))
num_generated_tokens += len(output.token_ids)
assert final_res.prompt_token_ids is not None
@@ -407,11 +461,10 @@ class ServingTokens(GenerateBaseServing):
spec_stats = build_spec_decoding_metrics(final_res)
if spec_stats is not None:
per_request_metrics = PerRequestMetrics(speculative_decoding=spec_stats)
response = GenerateResponse(
response_fields: dict[str, Any] = dict(
request_id=request_id,
created=created_time,
model=model_name,
choices=choices,
usage=usage,
prompt_logprobs=clamp_prompt_logprobs(final_res.prompt_logprobs),
prompt_token_id_logprobs=(
@@ -427,10 +480,15 @@ class ServingTokens(GenerateBaseServing):
kv_transfer_params=final_res.kv_transfer_params,
ec_transfer_params=final_res.ec_transfer_params,
)
response: GenerateTokensResponse | GenerateTextResponse
if text_mode:
response = GenerateTextResponse(choices=text_choices, **response_fields)
else:
response = GenerateTokensResponse(choices=tokens_choices, **response_fields)
# Log complete response if output logging is enabled
if self.enable_log_outputs and self.request_logger:
for choice in choices:
for choice in response.choices:
# Get the corresponding output token IDs
output_token_ids = None
if choice.index < len(final_res.outputs):
@@ -464,6 +522,8 @@ class ServingTokens(GenerateBaseServing):
num_cached_tokens = None
sampling_params: SamplingParams = request.sampling_params
last_res: RequestOutput | None = None
text_mode = request.output_mode == "text"
tokenizer = self._logprobs_tokenizer(text_mode)
include_usage, include_continuous_usage = should_include_usage(
request.stream_options, False
@@ -491,11 +551,21 @@ class ServingTokens(GenerateBaseServing):
finish_reason = output.finish_reason
self._raise_if_error(finish_reason, request_id)
# Still emit a terminal empty chunk while prompt metadata
# is pending, so zero-token completions deliver it.
if not delta_token_ids and (
finish_reason is None or prompt_token_ids is None
):
if text_mode:
# Text held back for stop string matching is flushed
# with the final output, and the abort output has no
# new token IDs, so text or a finish reason is enough.
emit = bool(
delta_token_ids or output.text or finish_reason is not None
)
else:
# Still emit a terminal empty chunk while prompt
# metadata is pending, so zero-token completions
# deliver it.
emit = bool(delta_token_ids) or (
finish_reason is not None and prompt_token_ids is not None
)
if not emit:
continue
if sampling_params.logprobs is not None:
@@ -505,6 +575,7 @@ class ServingTokens(GenerateBaseServing):
token_ids=delta_token_ids,
top_logprobs=out_logprobs,
num_output_top_logprobs=sampling_params.logprobs,
tokenizer=tokenizer,
)
else:
logprobs = None
@@ -518,20 +589,30 @@ class ServingTokens(GenerateBaseServing):
sampling_mask = None
if output.sampling_mask is not None:
sampling_mask = output.sampling_mask.token_ids
chunk = GenerateStreamResponse(
request_id=request_id,
choices=[
GenerateResponseStreamChoice(
index=i,
logprobs=logprobs,
finish_reason=finish_reason,
token_ids=as_list(delta_token_ids),
routed_experts=routed_experts_b64,
sampling_mask=sampling_mask,
)
],
choice_fields: dict[str, Any] = dict(
index=i,
logprobs=logprobs,
finish_reason=finish_reason,
token_ids=as_list(delta_token_ids),
routed_experts=routed_experts_b64,
sampling_mask=sampling_mask,
)
chunk: GenerateTokensStreamResponse | GenerateTextStreamResponse
if text_mode:
chunk = GenerateTextStreamResponse(
request_id=request_id,
choices=[
GenerateTextStreamChoice(
text=output.text, **choice_fields
)
],
)
else:
chunk = GenerateTokensStreamResponse(
request_id=request_id,
choices=[GenerateTokensStreamChoice(**choice_fields)],
)
if prompt_token_ids is not None:
chunk.prompt_token_ids = prompt_token_ids
chunk.mm_placeholders = request._response_mm_placeholders
@@ -571,12 +652,21 @@ class ServingTokens(GenerateBaseServing):
per_request_metrics = PerRequestMetrics(
speculative_decoding=spec_stats
)
final_chunk = GenerateStreamResponse(
request_id=request_id,
choices=[],
usage=final_usage_info,
metrics=per_request_metrics,
)
final_chunk: GenerateTokensStreamResponse | GenerateTextStreamResponse
if text_mode:
final_chunk = GenerateTextStreamResponse(
request_id=request_id,
choices=[],
usage=final_usage_info,
metrics=per_request_metrics,
)
else:
final_chunk = GenerateTokensStreamResponse(
request_id=request_id,
choices=[],
usage=final_usage_info,
metrics=per_request_metrics,
)
yield f"data: {final_chunk.model_dump_json(exclude_none=True)}\n\n"
request_metadata.final_usage_info = final_usage_info
@@ -591,37 +681,49 @@ class ServingTokens(GenerateBaseServing):
yield f"data: {data}\n\n"
yield "data: [DONE]\n\n"
def _logprobs_tokenizer(self, text_mode: bool) -> TokenizerLike | None:
"""Tokenizer that resolves logprob tokens, or None for placeholders.
``--return-tokens-as-token-ids`` keeps placeholders at every level, as
it does on ``/v1/completions``.
"""
if not text_mode or self.return_tokens_as_token_ids:
return None
return self.renderer.tokenizer
def _create_tokens_logprobs(
self,
token_ids: GenericSequence[int],
top_logprobs: GenericSequence[dict[int, Logprob] | None],
num_output_top_logprobs: int | None = None,
tokenizer: TokenizerLike | None = None,
) -> ChatCompletionLogProbs:
"""Create OpenAI-style logprobs."""
"""Create OpenAI-style logprobs.
Tokens are ``token_id:N`` placeholders. With a ``tokenizer`` they are
decoded strings and carry ``bytes``.
"""
logprobs_content: list[ChatCompletionLogProbsContent] = []
for i, token_id in enumerate(token_ids):
token = f"token_id:{token_id}"
step_top_logprobs = top_logprobs[i]
if step_top_logprobs is None or step_top_logprobs.get(token_id) is None:
token, token_bytes = _logprob_token(token_id, None, tokenizer)
logprobs_content.append(
ChatCompletionLogProbsContent(
token=token,
)
ChatCompletionLogProbsContent(token=token, bytes=token_bytes)
)
else:
step_token = step_top_logprobs[token_id]
token, token_bytes = _logprob_token(token_id, step_token, tokenizer)
logprobs_content.append(
ChatCompletionLogProbsContent(
token=token,
logprob=max(step_token.logprob, -9999.0),
bytes=token_bytes,
top_logprobs=[
ChatCompletionLogProb(
token=f"token_id:{token_id}",
logprob=max(logprob.logprob, -9999.0),
)
for i, (token_id, logprob) in enumerate(
_top_logprob(top_id, logprob, tokenizer)
for i, (top_id, logprob) in enumerate(
step_top_logprobs.items()
)
if num_output_top_logprobs is not None
vllm/renderers/online_derenderer.py
+9 -9
View File
@@ -28,8 +28,8 @@ from vllm.entrypoints.openai.completion.protocol import (
)
from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
DerenderStreamState,
GenerateResponse,
GenerateStreamResponse,
GenerateTokensResponse,
GenerateTokensStreamResponse,
)
from vllm.entrypoints.serve.engine.protocol import UsageInfo
from vllm.entrypoints.serve.utils.request_logger import RequestLogger
@@ -118,7 +118,7 @@ class OnlineDerenderer:
async def derender_chat(
self,
generate_response: GenerateResponse,
generate_response: GenerateTokensResponse,
chat_request: ChatCompletionRequest | None = None,
prompt_token_ids: list[int] | None = None,
) -> list[ChatCompletionResponseChoice]:
@@ -128,7 +128,7 @@ class OnlineDerenderer:
def _derender_chat(
self,
generate_response: GenerateResponse,
generate_response: GenerateTokensResponse,
chat_request: ChatCompletionRequest | None = None,
prompt_token_ids: list[int] | None = None,
) -> list[ChatCompletionResponseChoice]:
@@ -323,7 +323,7 @@ class OnlineDerenderer:
async def derender_chat_stream(
self,
model: str,
generate_chunk: GenerateStreamResponse,
generate_chunk: GenerateTokensStreamResponse,
state: DerenderStreamState | None = None,
chat_request: ChatCompletionRequest | None = None,
prompt_tokens: int | None = None,
@@ -472,7 +472,7 @@ class OnlineDerenderer:
self,
parser_cls: type[Parser],
model: str,
generate_chunk: GenerateStreamResponse,
generate_chunk: GenerateTokensStreamResponse,
state: DerenderStreamState,
chat_request: ChatCompletionRequest,
prompt_tokens: int | None,
@@ -682,7 +682,7 @@ class OnlineDerenderer:
async def derender_completion(
self,
generate_responses: list[GenerateResponse],
generate_responses: list[GenerateTokensResponse],
prompt_tokens: list[int] | None = None,
completion_request: CompletionRequest | None = None,
prompt_token_ids: list[list[int] | None] | None = None,
@@ -693,7 +693,7 @@ class OnlineDerenderer:
def _derender_completion(
self,
generate_responses: list[GenerateResponse],
generate_responses: list[GenerateTokensResponse],
prompt_tokens: list[int] | None = None,
completion_request: CompletionRequest | None = None,
prompt_token_ids: list[list[int] | None] | None = None,
@@ -761,7 +761,7 @@ class OnlineDerenderer:
async def derender_completion_stream(
self,
model: str,
generate_chunk: GenerateStreamResponse,
generate_chunk: GenerateTokensStreamResponse,
state: DerenderStreamState | None = None,
prompt_tokens: int | None = None,
completion_request: CompletionRequest | None = None,