diff --git a/docs/features/disagg_prefill.md b/docs/features/disagg_prefill.md index 013d57fa87..8fd67ca4c2 100644 --- a/docs/features/disagg_prefill.md +++ b/docs/features/disagg_prefill.md @@ -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. diff --git a/docs/serving/online_serving/README.md b/docs/serving/online_serving/README.md index a292a4a3a4..8e1a4f3398 100644 --- a/docs/serving/online_serving/README.md +++ b/docs/serving/online_serving/README.md @@ -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 diff --git a/docs/serving/online_serving/derenderer.md b/docs/serving/online_serving/derenderer.md index cf57f5d8b2..9585031312 100644 --- a/docs/serving/online_serving/derenderer.md +++ b/docs/serving/online_serving/derenderer.md @@ -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 diff --git a/docs/serving/online_serving/token_in_token_out.md b/docs/serving/online_serving/token_in_token_out.md new file mode 100644 index 0000000000..b4e0d465bd --- /dev/null +++ b/docs/serving/online_serving/token_in_token_out.md @@ -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). diff --git a/docs/usage/security.md b/docs/usage/security.md index 2b09ee5266..d350b6bdfd 100644 --- a/docs/usage/security.md +++ b/docs/usage/security.md @@ -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:** diff --git a/tests/entrypoints/scale_out/derender/test_derender_parity.py b/tests/entrypoints/scale_out/derender/test_derender_parity.py index a89f8d68db..66d3e83969 100644 --- a/tests/entrypoints/scale_out/derender/test_derender_parity.py +++ b/tests/entrypoints/scale_out/derender/test_derender_parity.py @@ -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 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) diff --git a/tests/entrypoints/scale_out/derender/test_derender_stream.py b/tests/entrypoints/scale_out/derender/test_derender_stream.py index 5862256835..0e4b168677 100644 --- a/tests/entrypoints/scale_out/derender/test_derender_stream.py +++ b/tests/entrypoints/scale_out/derender/test_derender_stream.py @@ -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( diff --git a/tests/entrypoints/scale_out/test_factories.py b/tests/entrypoints/scale_out/test_factories.py index 6d8335395e..ea96fff876 100644 --- a/tests/entrypoints/scale_out/test_factories.py +++ b/tests/entrypoints/scale_out/test_factories.py @@ -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 diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py index 53ff1dc598..6b6d732f07 100644 --- a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py +++ b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py @@ -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 diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_protocol.py b/tests/entrypoints/scale_out/token_in_token_out/test_protocol.py index e51530ce45..9894d09180 100644 --- a/tests/entrypoints/scale_out/token_in_token_out/test_protocol.py +++ b/tests/entrypoints/scale_out/token_in_token_out/test_protocol.py @@ -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] diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py b/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py index 18513f83dc..4a74448046 100644 --- a/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py +++ b/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py @@ -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 diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py b/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py index f42a406f41..c417a5fc2b 100644 --- a/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py +++ b/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py @@ -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>") diff --git a/vllm/entrypoints/launchers/cli_args.py b/vllm/entrypoints/launchers/cli_args.py index 4c1f121372..28fe79fd69 100644 --- a/vllm/entrypoints/launchers/cli_args.py +++ b/vllm/entrypoints/launchers/cli_args.py @@ -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. diff --git a/vllm/entrypoints/scale_out/derender/serving.py b/vllm/entrypoints/scale_out/derender/serving.py index 0783d3a78f..72ef434770 100644 --- a/vllm/entrypoints/scale_out/derender/serving.py +++ b/vllm/entrypoints/scale_out/derender/serving.py @@ -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. diff --git a/vllm/entrypoints/scale_out/token_in_token_out/api_router.py b/vllm/entrypoints/scale_out/token_in_token_out/api_router.py index 28677b6321..ee0e8c1623 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/api_router.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/api_router.py @@ -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) diff --git a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py index 60cffd07c2..31bd7e2eae 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py @@ -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.``. + """Whether the caller explicitly set `sampling_params.`. 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...` 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 - ````). `prompt_tokens` is a usage count and cannot serve this + ``). `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 = ( diff --git a/vllm/entrypoints/scale_out/token_in_token_out/serving.py b/vllm/entrypoints/scale_out/token_in_token_out/serving.py index 249ed48093..a30e519cb2 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/serving.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/serving.py @@ -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 diff --git a/vllm/renderers/online_derenderer.py b/vllm/renderers/online_derenderer.py index 0261401dd4..d221b32a8b 100644 --- a/vllm/renderers/online_derenderer.py +++ b/vllm/renderers/online_derenderer.py @@ -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,