From fcfacc3499b3f45978fc92a1b300f817cd17bb77 Mon Sep 17 00:00:00 2001 From: Martin Hickey Date: Fri, 2 Oct 2026 04:17:56 +0100 Subject: [PATCH] [Bugfix][Frontend] Seed derender detokenization from the prompt (#59046) Signed-off-by: Martin Hickey --- docs/serving/online_serving/derenderer.md | 12 +- requirements/common.txt | 2 +- .../scale_out/derender/test_derender.py | 135 ++++++ .../derender/test_derender_stream.py | 429 ++++++++++++++---- .../entrypoints/scale_out/derender/serving.py | 33 +- .../scale_out/token_in_token_out/protocol.py | 48 +- vllm/renderers/online_derenderer.py | 247 +++++++--- vllm/tokenizers/detokenizer_utils.py | 4 +- vllm/v1/engine/detokenizer.py | 13 +- 9 files changed, 760 insertions(+), 163 deletions(-) diff --git a/docs/serving/online_serving/derenderer.md b/docs/serving/online_serving/derenderer.md index e6d1221fd4..cf57f5d8b2 100644 --- a/docs/serving/online_serving/derenderer.md +++ b/docs/serving/online_serving/derenderer.md @@ -23,6 +23,10 @@ endpoints. The derender step needs more than the engine's `token_ids`. It also consumes the original `chat_request`/`completion_request` and `prompt_tokens` carried over from the render step (see [Request format](#request-format)) so the tool and reasoning parsers have the context they need. The chat streaming path additionally needs `prompt_token_ids` (see [Streaming cost](#streaming-cost)) when a parser is configured. +All derender endpoints also accept an optional `prompt_token_ids` (the `token_ids` of the `GenerateRequest` returned by `/render`). It's used to seed detokenization from the end of the prompt, the same way the engine does. Without it, the first output token is decoded as if it started a new sequence. On SentencePiece tokenizers (Llama-2, Phi-3) that drops its leading space compared to `/v1/chat/completions` and `/v1/completions`. If it's omitted, derender falls back to the generate response's `prompt_token_ids` (set when the `GenerateRequest` has `return_token_ids`), then to decoding without prompt context. + +Seeding only reads the last few IDs (7 today), so a suffix of the prompt is enough and a long prompt doesn't need to travel back to derender. The exception is streaming chat with a tool or reasoning parser configured. There the parser also reads `prompt_token_ids` to decide whether the prompt left reasoning open and it can look back to any point in the prompt. In this case, send the full list (see [Streaming cost](#streaming-cost)). + ## API Reference - Chat Completions Derender API (`/v1/chat/completions/derender`) @@ -96,6 +100,7 @@ The server keeps no state between calls. Everything the next call needs is in `s - **`n > 1` is N streams.** Generate interleaves chunks for different choices and tags each with its `index`. Keep one `stream_state` per index and send each chunk with the state for its index. Each index gets its own `role` delta on its first chunk. - **Send every chunk through, including the last ones.** The finish reason usually arrives with the final tokens. A chunk with a `finish_reason` and no tokens is accepted too and flushes any buffered tool call arguments. With `stream_options: {"include_usage": true}`, generate ends with a usage only chunk (`choices: []`). Send it through derender like any other chunk to get the usage chunk for the response. `usage.prompt_tokens` is the request's `prompt_tokens` if set, otherwise the generate chunk's. - **Send the same context on every call.** `chat_request` and `prompt_token_ids` aren't kept between calls, so they go with every chunk including the usage chunk. Both are required when a tool or reasoning parser is configured. `prompt_token_ids` is the `token_ids` of the `GenerateRequest` returned by `/render`. +- **Send `prompt_token_ids` on the first chunk, even without a parser.** It seeds detokenization so the first token's leading space matches the coupled endpoint (see [Request format](#request-format)). Later chunks already carry that context in `stream_state`. - **Don't forward `[DONE]`.** It marks the end of the generate stream and isn't a chunk. Streaming derender resolves `logprobs` on the plain detokenization path, including `token_id:N` placeholders, same as the non-streaming endpoints (see [Streaming state and logprobs](#streaming-state-and-logprobs)). The parser path doesn't resolve them yet and drops them. @@ -110,7 +115,7 @@ When a tool or reasoning parser is configured, parser internal state (buffered m - **Transport**: `output_token_ids` and `output_chunk_lens` round-trip in full in both directions on every call. `output_chunk_lens` has one entry per chunk, which is one per token without speculative decoding. This means O(n) bytes per chunk, O(n²) bytes over a full generation. Bounded by `max_model_len`. `prompt_token_ids` is sent in full on every call too and it isn't trimmed as `output_token_ids` grows. This means that for most of a stream it dominates the per chunk payload. A 100k token prompt with 1k tokens of output means `prompt_token_ids` is ~99% of the request body on every chunk. - **Compute**: replay is O(n) `parse_delta` calls per chunk (O(n²) per generation). `parse_delta` itself is O(n) for parsers that re-scan accumulated text (e.g. Hermes tool-call JSON, DeepSeek-R1 reasoning). The per-generation cost is O(n³) character work, not O(n²). This is a deliberately minimal first implementation with no caching layer. -- The parser path also requires `prompt_token_ids` so `parse_delta` can settle whether the prompt left reasoning open or not. Since parser state can't be carried across calls, it re-scans the full prompt once per chunk. +- The parser path also requires `prompt_token_ids` so `parse_delta` can settle whether the prompt left reasoning open or not. Since parser state can't be carried across calls, it re-scans the full prompt once per chunk. Send the full list and not a suffix because the last reasoning marker can be anywhere in the prompt. - Replay runs off the event loop on the renderer's executor (`renderer_num_workers`, default `1`). Size it for the expected number of concurrent parser configured streams. `output_token_ids` and `prompt_token_ids` are both bounded by `max_model_len` but callers streaming long reasoning traces through a parser configured model should expect materially more state transport and CPU cost than the plain detokenization path. @@ -153,7 +158,7 @@ with httpx.Client(timeout=60.0) as client: generate_request = client.post( f"{RENDER}/v1/chat/completions/render", json=chat_request ).json() - prompt_tokens = len(generate_request["token_ids"]) + prompt_token_ids = generate_request["token_ids"] # 2. Generate: token IDs -> token IDs (token-in / token-out engine) generate_response = client.post( @@ -166,7 +171,8 @@ with httpx.Client(timeout=60.0) as client: json={ "model": MODEL, "generate_response": generate_response, - "prompt_tokens": prompt_tokens, + "prompt_tokens": len(prompt_token_ids), + "prompt_token_ids": prompt_token_ids, "chat_request": chat_request, }, ).json() diff --git a/requirements/common.txt b/requirements/common.txt index 70bfa93253..8e483cc712 100644 --- a/requirements/common.txt +++ b/requirements/common.txt @@ -9,7 +9,7 @@ blake3 py-cpuinfo transformers >= 5.16.1 huggingface_hub >= 1.31.0 # Required for the `huggingface_hub.utils.httpx` re-export -tokenizers >= 0.21.1 # Required for fast incremental detokenization. +tokenizers >= 0.22.0 # Required for fast incremental detokenization. safetensors >= 0.6.2 # MXFP4/MXFP6 dtype support (F8_E8M0, F4) added in 0.6.0: https://github.com/huggingface/safetensors/pull/611 protobuf >= 5.29.6, !=6.30.*, !=6.31.*, !=6.32.*, !=6.33.0.*, !=6.33.1.*, !=6.33.2.*, !=6.33.3.*, !=6.33.4.*; platform_machine != "s390x" protobuf == 7.36.1; platform_machine == "s390x" diff --git a/tests/entrypoints/scale_out/derender/test_derender.py b/tests/entrypoints/scale_out/derender/test_derender.py index 80ecc0f49c..659aaed1aa 100644 --- a/tests/entrypoints/scale_out/derender/test_derender.py +++ b/tests/entrypoints/scale_out/derender/test_derender.py @@ -27,6 +27,11 @@ async def client(server): yield http_client +@pytest.fixture(scope="module") +def tokenizer(): + return get_tokenizer(MODEL_NAME) + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -372,6 +377,58 @@ async def test_derender_chat_model_omitted_resolves_served_name(client): assert response.json()["model"] == MODEL_NAME +@pytest.mark.asyncio +async def test_derender_chat_leading_space_seeded_from_prompt(client, tokenizer): + """prompt_token_ids keeps the first token's leading space on this + Metaspace tokenizer, matching the coupled path.""" + gen_req = await _render_chat(client) + prompt_token_ids = gen_req["token_ids"] + output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False) + + unseeded = await client.post( + "/v1/chat/completions/derender", + json={ + "model": MODEL_NAME, + "generate_response": _make_generate_response(output_ids), + }, + ) + seeded = await client.post( + "/v1/chat/completions/derender", + json={ + "model": MODEL_NAME, + "generate_response": _make_generate_response(output_ids), + "prompt_token_ids": prompt_token_ids, + }, + ) + assert unseeded.status_code == 200 + assert seeded.status_code == 200 + unseeded_content = unseeded.json()["choices"][0]["message"]["content"] + seeded_content = seeded.json()["choices"][0]["message"]["content"] + + assert not unseeded_content.startswith(" ") + assert seeded_content.startswith(" ") + assert seeded_content.lstrip(" ") == unseeded_content + + +@pytest.mark.asyncio +async def test_derender_chat_oversized_prompt_token_ids_rejected(client): + """prompt_token_ids longer than max_model_len returns 400.""" + gen_req = await _render_chat(client) + synthetic_ids = gen_req["token_ids"][:3] + oversized_prompt_ids = [42] * 1_000_000 + + response = await client.post( + "/v1/chat/completions/derender", + json={ + "model": MODEL_NAME, + "generate_response": _make_generate_response(synthetic_ids), + "prompt_token_ids": oversized_prompt_ids, + }, + ) + assert response.status_code == 400 + assert "max_model_len" in response.json()["error"]["message"] + + # --------------------------------------------------------------------------- # Completion derender tests # --------------------------------------------------------------------------- @@ -486,6 +543,84 @@ async def test_derender_completion_prompt_tokens_length_mismatch(client): assert response.status_code == 400 +@pytest.mark.asyncio +async def test_derender_completion_prompt_token_ids_length_mismatch(client): + """len(prompt_token_ids) != len(generate_responses) returns 400, the + same as the existing prompt_tokens length check.""" + gr1 = await _render_completion(client, "Hello") + ids1 = gr1["token_ids"][:3] + + response = await client.post( + "/v1/completions/derender", + json={ + "model": MODEL_NAME, + "generate_responses": [ + _make_completion_generate_response(ids1, gr1["request_id"]), + ], + "prompt_token_ids": [[1, 2], [3, 4]], + }, + ) + assert response.status_code == 400 + + +@pytest.mark.asyncio +async def test_derender_completion_leading_space_seeded_from_prompt(client, tokenizer): + """Completions counterpart of + test_derender_chat_leading_space_seeded_from_prompt.""" + gr1 = await _render_completion(client, "Hello world") + prompt_token_ids = gr1["token_ids"] + output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False) + + unseeded = await client.post( + "/v1/completions/derender", + json={ + "model": MODEL_NAME, + "generate_responses": [ + _make_completion_generate_response(output_ids, gr1["request_id"]), + ], + }, + ) + seeded = await client.post( + "/v1/completions/derender", + json={ + "model": MODEL_NAME, + "generate_responses": [ + _make_completion_generate_response(output_ids, gr1["request_id"]), + ], + "prompt_token_ids": [prompt_token_ids], + }, + ) + assert unseeded.status_code == 200 + assert seeded.status_code == 200 + unseeded_text = unseeded.json()["choices"][0]["text"] + seeded_text = seeded.json()["choices"][0]["text"] + + assert not unseeded_text.startswith(" ") + assert seeded_text.startswith(" ") + assert seeded_text.lstrip(" ") == unseeded_text + + +@pytest.mark.asyncio +async def test_derender_completion_oversized_prompt_token_ids_rejected(client): + """A prompt_token_ids entry longer than max_model_len returns 400.""" + gr1 = await _render_completion(client, "Hello") + ids1 = gr1["token_ids"][:3] + oversized_prompt_ids = [42] * 1_000_000 + + response = await client.post( + "/v1/completions/derender", + json={ + "model": MODEL_NAME, + "generate_responses": [ + _make_completion_generate_response(ids1, gr1["request_id"]), + ], + "prompt_token_ids": [oversized_prompt_ids], + }, + ) + assert response.status_code == 400 + assert "max_model_len" in response.json()["error"]["message"] + + @pytest.mark.asyncio async def test_derender_completion_empty_generate_responses(client): """Empty generate_responses list returns 400.""" diff --git a/tests/entrypoints/scale_out/derender/test_derender_stream.py b/tests/entrypoints/scale_out/derender/test_derender_stream.py index b1db1193a2..5862256835 100644 --- a/tests/entrypoints/scale_out/derender/test_derender_stream.py +++ b/tests/entrypoints/scale_out/derender/test_derender_stream.py @@ -20,11 +20,17 @@ Tests are split into two layers: import json import threading from concurrent.futures import ThreadPoolExecutor +from unittest.mock import AsyncMock, MagicMock +import httpx import pytest import pytest_asyncio +from pydantic import ValidationError +from tokenizers.decoders import DecodeStream +import vllm.renderers.online_derenderer as online_derenderer_module from tests.entrypoints.scale_out.derender.utils import stream_chat_derender +from tests.utils import RemoteLaunchRenderServer from vllm.entrypoints.generate.base.protocol import ( DeltaFunctionCall, DeltaMessage, @@ -32,7 +38,17 @@ from vllm.entrypoints.generate.base.protocol import ( PerRequestMetrics, SpeculativeDecodingMetrics, ) +from vllm.entrypoints.openai.chat_completion.protocol import ( + ChatCompletionRequest, + ChatCompletionStreamResponse, +) +from vllm.entrypoints.openai.completion.protocol import CompletionRequest +from vllm.entrypoints.scale_out.derender.serving import ServingDerender from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( + DerenderChatRequest, + DerenderChatStreamRequest, + DerenderCompletionRequest, + DerenderCompletionStreamRequest, DerenderStreamState, GenerateResponse, GenerateResponseChoice, @@ -41,6 +57,12 @@ from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( ) from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo from vllm.parser import Parser +from vllm.renderers.online_derenderer import ( + OnlineDerenderer, + _decode_params, + _seed_stream_state, +) +from vllm.tokenizers import get_tokenizer from vllm.utils import random_uuid MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM" @@ -170,6 +192,7 @@ def _make_stream_chunk( usage: dict | None = None, metrics: PerRequestMetrics | None = None, logprobs: dict | None = None, + prompt_token_ids: list[int] | None = None, ) -> GenerateStreamResponse: """Build a GenerateStreamResponse SSE chunk.""" return GenerateStreamResponse( @@ -184,6 +207,7 @@ def _make_stream_chunk( ], usage=UsageInfo(**usage) if usage else None, metrics=metrics, + prompt_token_ids=prompt_token_ids, ) @@ -228,18 +252,12 @@ def _make_usage_chunk( @pytest.fixture(scope="module") def tokenizer(): """Load the tiny tokenizer used across unit tests.""" - from vllm.tokenizers import get_tokenizer - return get_tokenizer(MODEL_NAME) @pytest.fixture(scope="module") def derenderer(tokenizer, request): """Construct a minimal OnlineDerenderer backed by a stub renderer.""" - from unittest.mock import MagicMock - - from vllm.renderers.online_derenderer import OnlineDerenderer - renderer = MagicMock() renderer.get_tokenizer.return_value = tokenizer @@ -273,10 +291,6 @@ def parsed_derenderer(tokenizer): `renderer._executor` must be a real `ThreadPoolExecutor`. `loop.run_in_executor` cannot submit work to a `MagicMock`. """ - from unittest.mock import MagicMock - - from vllm.renderers.online_derenderer import OnlineDerenderer - renderer = MagicMock() renderer.get_tokenizer.return_value = tokenizer renderer._executor = ThreadPoolExecutor(max_workers=2) @@ -433,6 +447,57 @@ class TestDetokenizeDelta: # Bounded by a small constant independent of len(token_ids) assert max_window <= 32 + def test_truncated_multibyte_char_is_held_back(self, derenderer, tokenizer): + """Output ending mid-character emits no U+FFFD: batch == stream == engine.""" + emoji_ids = tokenizer.encode("🌈", add_special_tokens=False) + byte_ids = emoji_ids[-4:] + assert len(byte_ids) == 4 + output_ids = byte_ids[:-1] + + batch, _ = derenderer._detokenize_delta( + tokenizer, output_ids, DerenderStreamState() + ) + streamed = self._chunked(derenderer, tokenizer, [[t] for t in output_ids]) + engine = _decode_stream_expected(tokenizer, [], output_ids) + + assert "�" not in batch + assert batch == streamed == engine + + full, _ = derenderer._detokenize_delta( + tokenizer, byte_ids, DerenderStreamState() + ) + assert full.endswith("🌈") + + @pytest.mark.parametrize( + ("skip", "spaces", "preserve", "fast", "expected"), + [ + (True, True, False, True, (True, True)), + (True, False, False, True, (True, True)), + (True, False, False, False, (True, False)), + (False, False, False, True, (False, False)), + (False, False, True, True, (False, False)), + (True, False, True, True, (False, False)), + (True, True, True, True, (False, True)), + ], + ) + def test_decode_params_mirror_engine( + self, monkeypatch, skip, spaces, preserve, fast, expected + ): + """Flags match the engine's fast or slow detokenizer for the tokenizer.""" + monkeypatch.setattr( + online_derenderer_module, "uses_fast_detokenizer", lambda _: fast + ) + request = CompletionRequest( + model=MODEL_NAME, + prompt="x", + skip_special_tokens=skip, + spaces_between_special_tokens=spaces, + ) + assert _decode_params(None, request, preserve_special=preserve) == expected + + def test_decode_params_default_without_request(self, tokenizer): + assert _decode_params(tokenizer, None) == (True, True) + def test_n_independent_streams_same_result(self, derenderer, tokenizer): """N parallel streams with the same token sequence give the same text.""" token_ids = tokenizer.encode("parallel streams")[:8] @@ -451,6 +516,242 @@ class TestDetokenizeDelta: assert results[0] == self._one_shot(tokenizer, token_ids) +# --------------------------------------------------------------------------- +# Prompt seeded detokenization (matches the engine's primed detokenizer) +# --------------------------------------------------------------------------- + + +def _decode_stream_expected( + tokenizer, prompt_ids: list[int], output_ids: list[int] +) -> str: + """Engine-equivalent decode via `DecodeStream`, primed by stepping the + prompt (`DecodeStream(ids=...)` needs tokenizers >= 0.22).""" + stream = DecodeStream(skip_special_tokens=True) + for tid in prompt_ids: + stream.step(tokenizer.backend_tokenizer, tid) + return "".join( + s + for tid in output_ids + if (s := stream.step(tokenizer.backend_tokenizer, tid)) is not None + ) + + +@pytest.fixture(scope="module") +def leading_space_ids(tokenizer): + """(prompt_ids, output_ids, expected) where seeded and unseeded decode + differ by the leading space.""" + prompt_ids = tokenizer.encode("[INST] Hi [/INST]", add_special_tokens=False) + output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False) + expected = _decode_stream_expected(tokenizer, prompt_ids, output_ids) + unseeded = tokenizer.decode(output_ids, skip_special_tokens=True) + assert expected.startswith(" ") and not unseeded.startswith(" "), ( + "fixture no longer reproduces the leading-space divergence; " + "pick a different prompt/output pair" + ) + return prompt_ids, output_ids, expected + + +class TestSeedStreamState: + """`_seed_stream_state`: pure helper building the initial decode state.""" + + def test_seeds_prev_tokens_from_prompt_tail(self, tokenizer, leading_space_ids): + prompt_ids, _, _ = leading_space_ids + state = _seed_stream_state(tokenizer, prompt_ids, skip_special_tokens=True) + assert state.prev_tokens + assert state.read_offset == len(state.prev_tokens) + + def test_none_prompt_token_ids_returns_empty_state(self, tokenizer): + assert ( + _seed_stream_state(tokenizer, None, skip_special_tokens=True) + == DerenderStreamState() + ) + + def test_empty_prompt_token_ids_returns_empty_state(self, tokenizer): + assert ( + _seed_stream_state(tokenizer, [], skip_special_tokens=True) + == DerenderStreamState() + ) + + +class TestPromptSeededLeadingSpace: + """Every decode entry point keeps the first token's leading space when + given prompt_token_ids.""" + + @pytest.mark.asyncio + async def test_chat_batch_seeded_via_explicit_prompt_token_ids( + self, derenderer, leading_space_ids + ): + prompt_ids, output_ids, expected = leading_space_ids + choices = await derenderer.derender_chat( + GenerateResponse( + request_id="t", + choices=[ + GenerateResponseChoice( + index=0, token_ids=output_ids, finish_reason="stop" + ) + ], + ), + prompt_token_ids=prompt_ids, + ) + assert choices[0].message.content == expected + + @pytest.mark.asyncio + async def test_chat_batch_seeded_via_generate_response_fallback( + self, derenderer, leading_space_ids + ): + prompt_ids, output_ids, expected = leading_space_ids + choices = await derenderer.derender_chat( + GenerateResponse( + request_id="t", + choices=[ + GenerateResponseChoice( + index=0, token_ids=output_ids, finish_reason="stop" + ) + ], + prompt_token_ids=prompt_ids, + ), + ) + assert choices[0].message.content == expected + + @pytest.mark.asyncio + async def test_chat_batch_explicit_takes_precedence_over_fallback( + self, derenderer, tokenizer, leading_space_ids + ): + 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( + request_id="t", + choices=[ + GenerateResponseChoice( + index=0, token_ids=output_ids, finish_reason="stop" + ) + ], + prompt_token_ids=wrong_ids, + ), + prompt_token_ids=prompt_ids, + ) + assert choices[0].message.content == expected + + @pytest.mark.asyncio + async def test_chat_batch_without_prompt_context_stays_unseeded( + self, derenderer, leading_space_ids + ): + _, output_ids, _ = leading_space_ids + choices = await derenderer.derender_chat( + GenerateResponse( + request_id="t", + choices=[ + GenerateResponseChoice( + index=0, token_ids=output_ids, finish_reason="stop" + ) + ], + ), + ) + assert not choices[0].message.content.startswith(" ") + + @pytest.mark.asyncio + async def test_completion_batch_seeded(self, derenderer, leading_space_ids): + prompt_ids, output_ids, expected = leading_space_ids + choices, _, _ = await derenderer.derender_completion( + [ + GenerateResponse( + request_id="t", + choices=[ + GenerateResponseChoice( + index=0, token_ids=output_ids, finish_reason="stop" + ) + ], + ) + ], + prompt_token_ids=[prompt_ids], + ) + assert choices[0].text == expected + + @pytest.mark.asyncio + @pytest.mark.parametrize("from_generate_chunk", [False, True]) + async def test_chat_stream_first_chunk_seeded( + self, derenderer, leading_space_ids, from_generate_chunk + ): + prompt_ids, output_ids, expected = leading_space_ids + mid = len(output_ids) // 2 + + chunk1, state = await derenderer.derender_chat_stream( + model=MODEL_NAME, + generate_chunk=_make_stream_chunk( + output_ids[:mid], + prompt_token_ids=prompt_ids if from_generate_chunk else None, + ), + prompt_token_ids=None if from_generate_chunk else prompt_ids, + ) + # Only needed on the first chunk; the carried state already holds + # the seeded window from here on. + chunk2, _ = await derenderer.derender_chat_stream( + model=MODEL_NAME, + generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"), + state=state, + ) + + streamed = (chunk1.choices[0].delta.content or "") + ( + chunk2.choices[0].delta.content or "" + ) + assert streamed == expected + + @pytest.mark.asyncio + @pytest.mark.parametrize("from_generate_chunk", [False, True]) + async def test_completion_stream_first_chunk_seeded( + self, derenderer, leading_space_ids, from_generate_chunk + ): + prompt_ids, output_ids, expected = leading_space_ids + mid = len(output_ids) // 2 + + chunk1, state = await derenderer.derender_completion_stream( + model=MODEL_NAME, + generate_chunk=_make_stream_chunk( + output_ids[:mid], + prompt_token_ids=prompt_ids if from_generate_chunk else None, + ), + prompt_token_ids=None if from_generate_chunk else prompt_ids, + ) + chunk2, _ = await derenderer.derender_completion_stream( + model=MODEL_NAME, + generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"), + state=state, + ) + + assert chunk1.choices[0].text + chunk2.choices[0].text == expected + + @pytest.mark.asyncio + async def test_parser_path_ephemeral_state_seeded( + self, parsed_derenderer, monkeypatch, leading_space_ids + ): + """The parser path's per-call detok window is seeded too, so + `parse_delta` sees the same text as coupled serving.""" + + class _EchoParser(Parser): + def parse_delta( + self, + delta_text, + delta_token_ids, + request, + prompt_token_ids=None, + *, + finished, + ): + return DeltaMessage(content=delta_text) if delta_text else None + + monkeypatch.setattr(parsed_derenderer, "parser", _EchoParser) + prompt_ids, output_ids, expected = leading_space_ids + + chunk, _ = await parsed_derenderer.derender_chat_stream( + model=MODEL_NAME, + generate_chunk=_make_stream_chunk(output_ids, finish_reason="stop"), + chat_request=_chat_request(), + prompt_token_ids=prompt_ids, + ) + assert chunk.choices[0].delta.content == expected + + @pytest.mark.asyncio @pytest.mark.parametrize( "stream_method", @@ -460,8 +761,6 @@ async def test_streaming_detokenization_runs_off_event_loop( derenderer, monkeypatch, stream_method ): """Streaming detokenization runs on the renderer executor.""" - import vllm.renderers.online_derenderer as online_derenderer_module - event_loop_thread_id = threading.get_ident() executor_thread_id = derenderer.renderer._executor.submit( threading.get_ident @@ -561,8 +860,6 @@ class TestDerenderCompletionStream: @pytest.mark.asyncio async def test_skip_special_tokens_threaded(self, derenderer, tokenizer): """completion_request.skip_special_tokens is honored (not hardcoded True).""" - from vllm.entrypoints.openai.completion.protocol import CompletionRequest - eos = tokenizer.eos_token_id if eos is None: pytest.skip("tokenizer has no eos token to exercise special stripping") @@ -789,8 +1086,6 @@ class TestDerenderChatStream: def _chat_request(**kwargs): - from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest - kwargs.setdefault("messages", [{"role": "user", "content": "hi"}]) kwargs.setdefault("model", MODEL_NAME) return ChatCompletionRequest(**kwargs) @@ -1072,8 +1367,6 @@ def harmony_encode(): @pytest.fixture(scope="module") def harmony_tokenizer(): pytest.importorskip("openai_harmony") - from vllm.tokenizers import get_tokenizer - return get_tokenizer(HARMONY_MODEL, trust_remote_code=True) @@ -1085,10 +1378,7 @@ def harmony_derenderer(harmony_tokenizer): with a real tokenizer and parser, so the replay path is exercised against Harmony's actual channel grammar rather than a stub. """ - from unittest.mock import MagicMock - from vllm.parser.harmony import HarmonyParser - from vllm.renderers.online_derenderer import OnlineDerenderer renderer = MagicMock() renderer.get_tokenizer.return_value = harmony_tokenizer @@ -1224,20 +1514,14 @@ class TestDerenderStreamStateValidation: """DerenderStreamState rejects malformed caller supplied offsets/lengths.""" def test_negative_prefix_offset_rejected(self): - from pydantic import ValidationError - with pytest.raises(ValidationError): DerenderStreamState(prefix_offset=-1) def test_negative_read_offset_rejected(self): - from pydantic import ValidationError - with pytest.raises(ValidationError): DerenderStreamState(read_offset=-1) def test_prev_tokens_over_cap_rejected(self): - from pydantic import ValidationError - with pytest.raises(ValidationError): DerenderStreamState(prev_tokens=["a"] * 1025) @@ -1246,14 +1530,10 @@ class TestDerenderStreamStateValidation: assert len(state.prev_tokens) == 1024 def test_output_chunk_lens_mismatch_rejected(self): - from pydantic import ValidationError - with pytest.raises(ValidationError): DerenderStreamState(output_token_ids=[1, 2, 3], output_chunk_lens=[1, 1]) def test_output_chunk_lens_zero_entry_rejected(self): - from pydantic import ValidationError - with pytest.raises(ValidationError): DerenderStreamState(output_token_ids=[1, 2], output_chunk_lens=[0, 2]) @@ -1264,14 +1544,40 @@ class TestDerenderStreamStateValidation: assert state.output_chunk_lens == [1, 2] +_GEN = {"request_id": "t", "choices": []} +_GEN_BAD = {**_GEN, "prompt_token_ids": [-1]} + + +@pytest.mark.parametrize( + "model_cls,payload", + [ + (DerenderChatRequest, {"generate_response": _GEN, "prompt_token_ids": [-1]}), + (DerenderChatRequest, {"generate_response": _GEN_BAD}), + ( + DerenderCompletionRequest, + {"generate_responses": [_GEN], "prompt_token_ids": [[-1]]}, + ), + ( + DerenderChatStreamRequest, + {"stream": True, "generate_chunk": _GEN, "prompt_token_ids": [-1]}, + ), + ( + DerenderCompletionStreamRequest, + {"stream": True, "generate_chunk": _GEN_BAD}, + ), + ], +) +def test_negative_prompt_token_ids_rejected(model_cls, payload): + """Negative prompt ids would reach convert_ids_to_tokens, so they get a + 400 at parse time like `GenerateRequest.token_ids` does.""" + with pytest.raises(ValidationError, match="prompt_token_ids"): + model_cls.model_validate(payload) + + class TestServingDerenderStreamErrorHandling: """Malformed stream_state must surface as 400 and not an unhandled 500.""" def _make_serving(self, side_effect: Exception): - from unittest.mock import AsyncMock, MagicMock - - from vllm.entrypoints.scale_out.derender.serving import ServingDerender - models = MagicMock() models.is_base_model.return_value = True models.model_config = MagicMock() @@ -1289,10 +1595,6 @@ class TestServingDerenderStreamErrorHandling: @pytest.mark.asyncio @pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")]) async def test_completion_stream_bad_state_returns_400(self, exc): - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderCompletionStreamRequest, - ) - serving = self._make_serving(exc) request = DerenderCompletionStreamRequest( stream=True, @@ -1307,10 +1609,6 @@ class TestServingDerenderStreamErrorHandling: @pytest.mark.asyncio @pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")]) async def test_chat_stream_bad_state_returns_400(self, exc): - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(exc) request = DerenderChatStreamRequest( stream=True, @@ -1329,13 +1627,6 @@ class TestServingDerenderStreamValidation: before touching the tokenizer).""" def _make_serving(self, *, parser_configured: bool, max_model_len: int = 100_000): - from unittest.mock import AsyncMock, MagicMock - - from vllm.entrypoints.openai.chat_completion.protocol import ( - ChatCompletionStreamResponse, - ) - from vllm.entrypoints.scale_out.derender.serving import ServingDerender - models = MagicMock() models.is_base_model.return_value = True models.model_config = MagicMock() @@ -1353,10 +1644,6 @@ class TestServingDerenderStreamValidation: @pytest.mark.asyncio async def test_missing_chat_request_with_parser_rejected(self): - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(parser_configured=True) request = DerenderChatStreamRequest( stream=True, @@ -1371,10 +1658,6 @@ class TestServingDerenderStreamValidation: @pytest.mark.asyncio async def test_missing_chat_request_without_parser_ok(self): - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(parser_configured=False) request = DerenderChatStreamRequest( stream=True, @@ -1390,10 +1673,6 @@ class TestServingDerenderStreamValidation: the same way it rejects a missing chat_request. Without it, parse_delta cannot tell whether the prompt left reasoning open and would misclassify reasoning content as plain content.""" - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(parser_configured=True) request = DerenderChatStreamRequest( stream=True, @@ -1409,10 +1688,6 @@ class TestServingDerenderStreamValidation: @pytest.mark.asyncio async def test_prompt_token_ids_present_with_parser_ok(self): - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(parser_configured=True) request = DerenderChatStreamRequest( stream=True, @@ -1426,10 +1701,6 @@ class TestServingDerenderStreamValidation: @pytest.mark.asyncio async def test_oversized_output_token_ids_rejected(self): - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(parser_configured=False, max_model_len=4) request = DerenderChatStreamRequest( stream=True, @@ -1451,10 +1722,6 @@ class TestServingDerenderStreamValidation: parser configured deployment rescans it in full on every chunk (is_reasoning_end / adjust_initial_state_from_prompt), so it must be bounded the same way output_token_ids is.""" - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(parser_configured=True, max_model_len=4) request = DerenderChatStreamRequest( stream=True, @@ -1475,10 +1742,6 @@ class TestServingDerenderStreamValidation: DerenderStreamState is threaded through every choice). The check could never fire since derender_chat_stream itself rejects anything above 1 first.""" - from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( - DerenderChatStreamRequest, - ) - serving = self._make_serving(parser_configured=False) two_choices = GenerateStreamResponse( request_id="t", @@ -1503,16 +1766,12 @@ class TestServingDerenderStreamValidation: @pytest.fixture(scope="module") def server(): - from tests.utils import RemoteLaunchRenderServer - with RemoteLaunchRenderServer(MODEL_NAME, []) as remote_server: yield remote_server @pytest_asyncio.fixture async def client(server): - import httpx - async with httpx.AsyncClient( base_url=server.url_for(""), timeout=30.0 ) as http_client: @@ -1776,8 +2035,6 @@ _PARSER_TOOLS = [ @pytest.fixture(scope="module") def parser_server(): - from tests.utils import RemoteLaunchRenderServer - args = [ "--enable-auto-tool-choice", "--tool-call-parser", @@ -1791,8 +2048,6 @@ def parser_server(): @pytest_asyncio.fixture async def parser_client(parser_server): - import httpx - async with httpx.AsyncClient( base_url=parser_server.url_for(""), timeout=60.0 ) as http_client: @@ -1801,8 +2056,6 @@ async def parser_client(parser_server): @pytest.fixture(scope="module") def parser_tokenizer(): - from vllm.tokenizers import get_tokenizer - return get_tokenizer(PARSER_MODEL) diff --git a/vllm/entrypoints/scale_out/derender/serving.py b/vllm/entrypoints/scale_out/derender/serving.py index 5da7366403..0783d3a78f 100644 --- a/vllm/entrypoints/scale_out/derender/serving.py +++ b/vllm/entrypoints/scale_out/derender/serving.py @@ -117,6 +117,20 @@ class ServingDerender(BaseServing): return None + def _validate_prompt_token_ids( + self, + prompt_token_ids: list[list[int] | None], + ) -> ErrorResponse | None: + """Reject caller-supplied prompt_token_ids longer than max_model_len.""" + max_model_len = self.model_config.max_model_len + for ids in prompt_token_ids: + if ids is not None and len(ids) > max_model_len: + return self.create_error_response( + f"prompt_token_ids length ({len(ids)}) exceeds " + f"max_model_len ({max_model_len})." + ) + return None + async def derender_chat_response( self, request: DerenderChatRequest, @@ -138,6 +152,10 @@ class ServingDerender(BaseServing): if bounds_error is not None: return bounds_error + prompt_ids_error = self._validate_prompt_token_ids([request.prompt_token_ids]) + if prompt_ids_error is not None: + return prompt_ids_error + if self.online_derenderer.parser is not None and request.chat_request is None: return self.create_error_response( "chat_request is required when a tool or reasoning parser is " @@ -147,7 +165,9 @@ class ServingDerender(BaseServing): try: choices = await self.online_derenderer.derender_chat( - request.generate_response, request.chat_request + request.generate_response, + request.chat_request, + request.prompt_token_ids, ) except ValueError as exc: return self.create_error_response(str(exc)) @@ -203,6 +223,11 @@ class ServingDerender(BaseServing): if bounds_error is not None: return bounds_error + if request.prompt_token_ids is not None: + prompt_ids_error = self._validate_prompt_token_ids(request.prompt_token_ids) + if prompt_ids_error is not None: + return prompt_ids_error + ( choices, total_prompt_tokens, @@ -211,6 +236,7 @@ class ServingDerender(BaseServing): request.generate_responses, request.prompt_tokens, completion_request=request.completion_request, + prompt_token_ids=request.prompt_token_ids, ) first = request.generate_responses[0] @@ -367,6 +393,10 @@ class ServingDerender(BaseServing): if error_check_ret is not None: return error_check_ret + prompt_ids_error = self._validate_prompt_token_ids([request.prompt_token_ids]) + if prompt_ids_error is not None: + return prompt_ids_error + model_name = request.model or self.models.model_name() try: ( @@ -378,6 +408,7 @@ class ServingDerender(BaseServing): state=request.stream_state, prompt_tokens=request.prompt_tokens, completion_request=request.completion_request, + prompt_token_ids=request.prompt_token_ids, ) except ValueError as exc: return self.create_error_response(str(exc)) 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 94604b013f..60cffd07c2 100644 --- a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py +++ b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py @@ -5,6 +5,7 @@ from typing import Annotated, Any, Literal, TypeAlias from pydantic import ( BaseModel, Field, + NonNegativeInt, PrivateAttr, field_validator, model_validator, @@ -362,7 +363,7 @@ class GenerateStreamResponse(BaseModel): ) choices: list[GenerateResponseStreamChoice] usage: UsageInfo | None = Field(default=None) - prompt_token_ids: list[int] | None = None + prompt_token_ids: list[NonNegativeInt] | None = None mm_placeholders: dict[str, list[PlaceholderRangeInfo]] | None = None metrics: PerRequestMetrics | None = None @@ -382,7 +383,7 @@ class GenerateResponse(BaseModel): usage: UsageInfo | None = Field(default=None) prompt_logprobs: list[dict[int, Logprob] | None] | None = None prompt_token_id_logprobs: str | None = None - prompt_token_ids: list[int] | None = None + prompt_token_ids: list[NonNegativeInt] | None = None mm_placeholders: dict[str, list[PlaceholderRangeInfo]] | None = None metrics: PerRequestMetrics | None = None @@ -421,6 +422,15 @@ class DerenderChatRequest(BaseModel): len(GenerateRequest.token_ids) from the render step. """ + prompt_token_ids: list[NonNegativeInt] | None = None + """Prompt token IDs (`GenerateRequest.token_ids` from /render). Seeds + detokenization from the prompt tail so the first output token keeps its + leading space on SentencePiece tokenizers. Falls back to + `generate_response.prompt_token_ids`, then to unseeded decoding. + + Only the last few IDs are read, so a suffix of the prompt is enough. + """ + chat_request: ChatCompletionRequest | None = None """The original (post-adjust_request) ChatCompletionRequest from /render. @@ -455,6 +465,14 @@ class DerenderCompletionRequest(BaseModel): If provided, len(prompt_tokens) must equal len(generate_responses). """ + prompt_token_ids: list[list[NonNegativeInt] | None] | None = None + """One prompt token ID list per response, used to seed detokenization. + See `DerenderChatRequest.prompt_token_ids`. + + If provided, len(prompt_token_ids) must equal len(generate_responses). + A `None` entry falls back to `generate_responses[i].prompt_token_ids`. + """ + completion_request: CompletionRequest | None = None """The original (post-adjust_request) CompletionRequest from /render. @@ -472,6 +490,13 @@ class DerenderCompletionRequest(BaseModel): f"prompt_tokens length ({len(self.prompt_tokens)}) must equal " f"generate_responses length ({len(self.generate_responses)})" ) + if self.prompt_token_ids is not None and len(self.prompt_token_ids) != len( + self.generate_responses + ): + raise ValueError( + f"prompt_token_ids length ({len(self.prompt_token_ids)}) must " + f"equal generate_responses length ({len(self.generate_responses)})" + ) return self @@ -627,7 +652,7 @@ class DerenderChatStreamRequest(BaseModel): prompt_tokens: int | None = None """Prompt token count for usage. Forwarded from the render step.""" - prompt_token_ids: list[int] | None = None + 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 @@ -636,8 +661,15 @@ class DerenderChatStreamRequest(BaseModel): Rejected with a 400 (by `ServingDerender`) when a tool or reasoning parser is configured and this is omitted. Without it, `parse_delta` cannot tell whether the prompt left reasoning open and would silently - misclassify reasoning content as plain content. Unused on the plain - detokenization path. + misclassify reasoning content as plain content. On all paths it also + seeds detokenization on the first chunk, falling back to + `generate_chunk.prompt_token_ids` when omitted (see + `DerenderChatRequest.prompt_token_ids`). + + With a parser configured, send the full list. Parsers look back to the + last reasoning marker which can be anywhere in the prompt, so a suffix + can flip the initial reasoning state. Without a parser, a suffix is + enough. """ chat_request: ChatCompletionRequest | None = None @@ -666,6 +698,12 @@ class DerenderCompletionStreamRequest(BaseModel): prompt_tokens: int | None = None """Prompt token count for usage.""" + prompt_token_ids: list[NonNegativeInt] | None = None + """Prompt token IDs, used on the first chunk to seed detokenization. + Falls back to `generate_chunk.prompt_token_ids`. See + `DerenderChatRequest.prompt_token_ids`. + """ + completion_request: CompletionRequest | None = None """The original (post adjust_request) CompletionRequest from /render.""" # --8<-- [end:derender-completion-stream-request] diff --git a/vllm/renderers/online_derenderer.py b/vllm/renderers/online_derenderer.py index c1f175798d..0261401dd4 100644 --- a/vllm/renderers/online_derenderer.py +++ b/vllm/renderers/online_derenderer.py @@ -40,9 +40,14 @@ from vllm.logger import init_logger from vllm.parser import Parser, ParserManager from vllm.renderers import BaseRenderer from vllm.tokenizers import TokenizerLike -from vllm.tokenizers.detokenizer_utils import detokenize_incrementally +from vllm.tokenizers.detokenizer_utils import ( + convert_prompt_ids_to_tokens, + detokenize_incrementally, + get_leading_space_marker, +) from vllm.utils import random_uuid from vllm.utils.async_utils import make_async +from vllm.v1.engine.detokenizer import uses_fast_detokenizer logger = init_logger(__name__) @@ -115,17 +120,37 @@ class OnlineDerenderer: self, generate_response: GenerateResponse, chat_request: ChatCompletionRequest | None = None, + prompt_token_ids: list[int] | None = None, ) -> list[ChatCompletionResponseChoice]: - return await self._derender_chat_async(generate_response, chat_request) + return await self._derender_chat_async( + generate_response, chat_request, prompt_token_ids + ) def _derender_chat( self, generate_response: GenerateResponse, chat_request: ChatCompletionRequest | None = None, + prompt_token_ids: list[int] | None = None, ) -> list[ChatCompletionResponseChoice]: tokenizer = self.renderer.get_tokenizer() choices: list[ChatCompletionResponseChoice] = [] + has_parser = self.parser is not None and chat_request is not None + skip_special, spaces_between = _decode_params( + tokenizer, chat_request, preserve_special=has_parser + ) + seed_ids = ( + prompt_token_ids + if prompt_token_ids is not None + else generate_response.prompt_token_ids + ) + # Choices continue the same prompt, so they share one seeded window. + seed_state = _seed_stream_state( + tokenizer, + seed_ids, + skip_special_tokens=skip_special, + ) + for choice in generate_response.choices: if not choice.token_ids: raise ValueError(f"choice {choice.index} has empty or null token_ids") @@ -136,12 +161,17 @@ class OnlineDerenderer: else None ) - if self.parser is not None and chat_request is not None: + if has_parser: + assert self.parser is not None and chat_request is not None # Parser path: decode with special tokens preserved # so the parser can see markers like , # , or Harmony channel tokens. - decoded_text = tokenizer.decode( - choice.token_ids, skip_special_tokens=False + decoded_text, _ = self._detokenize_delta( + tokenizer, + choice.token_ids, + seed_state, + skip_special_tokens=skip_special, + spaces_between_special_tokens=spaces_between, ) chat_template_kwargs: dict[str, Any] = {} @@ -201,13 +231,12 @@ class OnlineDerenderer: else: # No parser: plain detokenization honouring the request's # skip_special_tokens (default True when no request was given). - skip_special = ( - chat_request.skip_special_tokens - if chat_request is not None - else True - ) - decoded_text = tokenizer.decode( - choice.token_ids, skip_special_tokens=skip_special + decoded_text, _ = self._detokenize_delta( + tokenizer, + choice.token_ids, + seed_state, + skip_special_tokens=skip_special, + spaces_between_special_tokens=spaces_between, ) message = ChatMessage(role="assistant", content=decoded_text) @@ -239,7 +268,7 @@ class OnlineDerenderer: (tracked by ``read_offset``). The delta tokens are fed straight onto it. The window is bounded. ``detokenize_incrementally`` never reads before - ``prefix_offset``, so after processing we trim ``prev_tokens`` to that + ``prefix_offset``, so after each token we trim ``prev_tokens`` to that tail and rebase the offsets to it. State transport therefore stays O(window) per chunk instead of re-sending the full token history. @@ -273,18 +302,20 @@ class OnlineDerenderer: skip_special_tokens=skip_special_tokens, spaces_between_special_tokens=spaces_between_special_tokens, ) - prev_tokens = prev_tokens + new_toks + # Trim to the tail still readable by detokenize_incrementally + # (everything before prefix_offset is dead) and rebase the + # offsets, so the window stays bounded within a long batch + # decode as well as across chunks. + prev_tokens = (prev_tokens + new_toks)[prefix_offset:] + read_offset -= prefix_offset + prefix_offset = 0 text_parts.append(text) - # Trim to the tail still readable by detokenize_incrementally - # (everything before prefix_offset is dead) and rebase the offsets so - # the carried window stays bounded regardless of generation length. - trimmed = prev_tokens[prefix_offset:] updated_state = state.model_copy( update={ - "prev_tokens": trimmed, - "prefix_offset": 0, - "read_offset": read_offset - prefix_offset, + "prev_tokens": prev_tokens, + "prefix_offset": prefix_offset, + "read_offset": read_offset, } ) return "".join(text_parts), updated_state @@ -315,20 +346,16 @@ class OnlineDerenderer: (validated by the caller — see `ServingDerender`) because plain detokenization would leak raw parser markup into `content`. prompt_tokens: Prompt token count for the usage chunk. - prompt_token_ids: Prompt token IDs. Parser path only. Lets - `parse_delta` settle its initial reasoning state. Required - when a reasoning or tool parser is configured (validated by - the caller — see ``ServingDerender``). Without it reasoning - left open by the prompt would be misclassified as content. + prompt_token_ids: Prompt token IDs. Required on the parser path + (validated by the caller) and optional otherwise, falling + back to ``generate_chunk.prompt_token_ids``. See + `DerenderChatStreamRequest.prompt_token_ids`. Returns: (chunk, updated_state) — the derendered SSE chunk and the state the client must pass to the next call. """ - if state is None: - state = DerenderStreamState() - # A single DerenderStreamState is threaded through every choice in # this chunk. Correct only when there is at most one choice per SSE # event (n=1, one call per index), as the streaming derender @@ -353,23 +380,34 @@ class OnlineDerenderer: parser_cls, model, generate_chunk, - state, + state if state is not None else DerenderStreamState(), chat_request, prompt_tokens, prompt_token_ids, ) tokenizer = self.renderer.get_tokenizer() - skip_special = ( - chat_request.skip_special_tokens if chat_request is not None else True - ) + skip_special, spaces_between = _decode_params(tokenizer, chat_request) + # Seed on the first chunk only. A carried state already has the prompt. + if state is None: + state = _seed_stream_state( + tokenizer, + prompt_token_ids + if prompt_token_ids is not None + else generate_chunk.prompt_token_ids, + skip_special_tokens=skip_special, + ) stream_choices: list[ChatCompletionResponseStreamChoice] = [] updated_state = state for choice in generate_chunk.choices: delta_tids = choice.token_ids or [] new_text, updated_state = await self._detokenize_delta_async( - tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special + tokenizer, + delta_tids, + updated_state, + skip_special_tokens=skip_special, + spaces_between_special_tokens=spaces_between, ) # NOTE: parser-configured servers dispatch to @@ -454,7 +492,8 @@ class OnlineDerenderer: `state.output_chunk_lens`, so the rebuilt parser state matches standard serving, which calls `parse_delta` once per engine step. Text comes from a fresh incremental detokenizer with special tokens - preserved (``skip_special_tokens=False``). + preserved (``skip_special_tokens=False``), seeded from the prompt + tail on every call since replay starts from scratch. """ tokenizer = self.renderer.get_tokenizer() @@ -480,7 +519,12 @@ class OnlineDerenderer: # across both the replay and current chunk phases (via `nonlocal`) # so multi-byte characters split across that boundary still decode # correctly. Discarded once the call returns. - detok_state = DerenderStreamState() + skip_special, spaces_between = _decode_params( + tokenizer, chat_request, preserve_special=True + ) + detok_state = _seed_stream_state( + tokenizer, prompt_token_ids, skip_special_tokens=skip_special + ) def _replay(token_ids: list[int], chunk_lens: list[int]) -> None: """Replay prior chunks through `parse_delta` to rebuild parser @@ -493,7 +537,11 @@ class OnlineDerenderer: chunk = token_ids[start : start + chunk_len] start += chunk_len text, detok_state = self._detokenize_delta( - tokenizer, chunk, detok_state, skip_special_tokens=False + tokenizer, + chunk, + detok_state, + skip_special_tokens=skip_special, + spaces_between_special_tokens=spaces_between, ) parser.parse_delta( text, @@ -530,7 +578,11 @@ class OnlineDerenderer: # granularity), not one per token. See the granularity note # in this method's docstring. text, detok_state = self._detokenize_delta( - tokenizer, delta_tids, detok_state, skip_special_tokens=False + tokenizer, + delta_tids, + detok_state, + skip_special_tokens=skip_special, + spaces_between_special_tokens=spaces_between, ) delta_message = parser.parse_delta( text, @@ -633,9 +685,10 @@ class OnlineDerenderer: generate_responses: list[GenerateResponse], prompt_tokens: list[int] | None = None, completion_request: CompletionRequest | None = None, + prompt_token_ids: list[list[int] | None] | None = None, ) -> tuple[list[CompletionResponseChoice], int, int]: return await self._derender_completion_async( - generate_responses, prompt_tokens, completion_request + generate_responses, prompt_tokens, completion_request, prompt_token_ids ) def _derender_completion( @@ -643,24 +696,34 @@ class OnlineDerenderer: generate_responses: list[GenerateResponse], prompt_tokens: list[int] | None = None, completion_request: CompletionRequest | None = None, + prompt_token_ids: list[list[int] | None] | None = None, ) -> tuple[list[CompletionResponseChoice], int, int]: n = len(generate_responses) prompt_tokens_list: list[int] = ( prompt_tokens if prompt_tokens is not None else [0] * n ) - - skip_special = ( - completion_request.skip_special_tokens - if completion_request is not None - else True + prompt_token_ids_list: list[list[int] | None] = ( + prompt_token_ids if prompt_token_ids is not None else [None] * n ) + tokenizer = self.renderer.get_tokenizer() + skip_special, spaces_between = _decode_params(tokenizer, completion_request) choices: list[CompletionResponseChoice] = [] total_prompt_tokens = 0 total_completion_tokens = 0 index = 0 - for gen, pt in zip(generate_responses, prompt_tokens_list): + for gen, pt, seed_ids_override in zip( + generate_responses, prompt_tokens_list, prompt_token_ids_list + ): + seed_ids = ( + seed_ids_override + if seed_ids_override is not None + else gen.prompt_token_ids + ) + seed_state = _seed_stream_state( + tokenizer, seed_ids, skip_special_tokens=skip_special + ) for choice in gen.choices: if not choice.token_ids: raise ValueError( @@ -668,8 +731,12 @@ class OnlineDerenderer: "has empty or null token_ids" ) - decoded_text = tokenizer.decode( - choice.token_ids, skip_special_tokens=skip_special + decoded_text, _ = self._detokenize_delta( + tokenizer, + choice.token_ids, + seed_state, + skip_special_tokens=skip_special, + spaces_between_special_tokens=spaces_between, ) completion_logprobs = None if choice.logprobs is not None: @@ -698,6 +765,7 @@ class OnlineDerenderer: state: DerenderStreamState | None = None, prompt_tokens: int | None = None, completion_request: CompletionRequest | None = None, + prompt_token_ids: list[int] | None = None, ) -> tuple[CompletionStreamResponse, DerenderStreamState]: """Process one GenerateStreamResponse chunk for streaming completions. @@ -716,14 +784,13 @@ class OnlineDerenderer: prompt_tokens: Prompt token count for usage (from the render step). completion_request: Original CompletionRequest from ``/render``; supplies ``skip_special_tokens``. + prompt_token_ids: Seeds the first chunk's decode. Falls back to + ``generate_chunk.prompt_token_ids``. Returns: (chunk, updated_state) — the derendered chunk and updated state. """ - if state is None: - state = DerenderStreamState() - # See the equivalent check in derender_chat_stream: a single # DerenderStreamState is threaded through every choice in this # chunk, so more than one choice per chunk would corrupt the @@ -734,18 +801,27 @@ class OnlineDerenderer: ) tokenizer = self.renderer.get_tokenizer() - skip_special = ( - completion_request.skip_special_tokens - if completion_request is not None - else True - ) + skip_special, spaces_between = _decode_params(tokenizer, completion_request) + # Seed on the first chunk only. A carried state already has the prompt. + if state is None: + state = _seed_stream_state( + tokenizer, + prompt_token_ids + if prompt_token_ids is not None + else generate_chunk.prompt_token_ids, + skip_special_tokens=skip_special, + ) stream_choices: list[CompletionResponseStreamChoice] = [] updated_state = state for choice in generate_chunk.choices: delta_tids = choice.token_ids or [] new_text, updated_state = await self._detokenize_delta_async( - tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special + tokenizer, + delta_tids, + updated_state, + skip_special_tokens=skip_special, + spaces_between_special_tokens=spaces_between, ) completion_logprobs = None @@ -810,6 +886,65 @@ def _logprob_context_tail( return (list(context_token_ids) + list(delta_token_ids))[-_LOGPROB_CONTEXT_WINDOW:] +def _decode_params( + tokenizer: TokenizerLike, + request: ChatCompletionRequest | CompletionRequest | None, + preserve_special: bool = False, +) -> tuple[bool, bool]: + """Derive `(skip_special_tokens, spaces_between_special_tokens)` the way + the engine does (`IncrementalDetokenizer`). + + Args: + tokenizer: Picks the engine's fast or slow detokenizer rules. + request: The original request, if the caller supplied one. + preserve_special: Keep special tokens so a parser can see markers. + The serving side does this via `adjust_request`. + + """ + skip_special = ( + False + if preserve_special + else (request.skip_special_tokens if request is not None else True) + ) + spaces_between = ( + request.spaces_between_special_tokens if request is not None else True + ) + # Only the fast detokenizer forces spaces on when skipping special tokens. + if skip_special and uses_fast_detokenizer(tokenizer): + spaces_between = True + return skip_special, spaces_between + + +def _seed_stream_state( + tokenizer: TokenizerLike, + prompt_token_ids: list[int] | None, + skip_special_tokens: bool, +) -> DerenderStreamState: + """Build the initial decode state from the prompt tail, the same way the + engine primes its incremental detokenizer. Without it, Metaspace + tokenizers drop the first output token's leading space. + + Returns an empty state when `prompt_token_ids` is empty or omitted. + """ + if not prompt_token_ids: + if prompt_token_ids is None and get_leading_space_marker(tokenizer) is not None: + logger.warning_once( + "derender got no prompt_token_ids, so the first output token " + "may lose its leading space compared to the coupled endpoint. " + "Pass prompt_token_ids from the render step to avoid this." + ) + return DerenderStreamState() + + prev_tokens, prefix_offset, read_offset = convert_prompt_ids_to_tokens( + tokenizer, prompt_token_ids, skip_special_tokens=skip_special_tokens + ) + return DerenderStreamState( + prev_tokens=prev_tokens, + prefix_offset=prefix_offset, + read_offset=read_offset, + ) + + def _parse_token_id_placeholder(token: str) -> int | None: """Extract token ID from a 'token_id:N' placeholder string.""" if not token.startswith("token_id:"): diff --git a/vllm/tokenizers/detokenizer_utils.py b/vllm/tokenizers/detokenizer_utils.py index 35e73bc410..ddaf512442 100644 --- a/vllm/tokenizers/detokenizer_utils.py +++ b/vllm/tokenizers/detokenizer_utils.py @@ -63,7 +63,7 @@ _CACHED_MARKER_KEY = "_vllm_space_marker_cache" _NOT_CACHED = "__not_computed__" -def _get_leading_space_marker(tokenizer: TokenizerLike) -> str | None: +def get_leading_space_marker(tokenizer: TokenizerLike) -> str | None: """Read the space marker from the tokenizer's pre_tokenizer config. Only Metaspace pre_tokenizers (used by SentencePiece-based models like @@ -160,7 +160,7 @@ def convert_ids_list_to_tokens( """ if not token_ids: return [] - marker = _get_leading_space_marker(tokenizer) + marker = get_leading_space_marker(tokenizer) if marker is None: return [tokenizer.decode([tid]) or "" for tid in token_ids] raw_tokens = tokenizer.convert_ids_to_tokens(token_ids) diff --git a/vllm/v1/engine/detokenizer.py b/vllm/v1/engine/detokenizer.py index 9658c030ef..c27e01895c 100644 --- a/vllm/v1/engine/detokenizer.py +++ b/vllm/v1/engine/detokenizer.py @@ -3,9 +3,7 @@ import sys from abc import ABC, abstractmethod -import tokenizers import tokenizers.decoders -from packaging import version from tokenizers import Tokenizer from transformers import TokenizersBackend @@ -20,14 +18,15 @@ from vllm.v1.engine import EngineCoreRequest logger = init_logger(__name__) -# Only tokenizers >= 0.22.0 supports DecodeStream with native prefill -# (ids parameter) used for FastIncrementalDetokenizer. -USE_FAST_DETOKENIZER = version.parse(tokenizers.__version__) >= version.parse("0.22.0") - # Error string from https://github.com/huggingface/tokenizers/blob/909fdde2a4ffedd9295206f705eb612be2a91b12/tokenizers/src/tokenizer/mod.rs#L1042 INVALID_PREFIX_ERR_MSG = "Invalid prefix encountered" +def uses_fast_detokenizer(tokenizer: TokenizerLike) -> bool: + """Whether the engine detokenizes with `FastIncrementalDetokenizer`.""" + return isinstance(tokenizer, TokenizersBackend) + + class IncrementalDetokenizer: def __init__(self): self.token_ids: list[int] = [] @@ -58,7 +57,7 @@ class IncrementalDetokenizer: # No tokenizer => skipping detokenization. return IncrementalDetokenizer() - if USE_FAST_DETOKENIZER and isinstance(tokenizer, TokenizersBackend): + if uses_fast_detokenizer(tokenizer): # Fast tokenizer => use tokenizers library DecodeStream. return FastIncrementalDetokenizer(tokenizer, request)