[Bugfix][Frontend] Seed derender detokenization from the prompt (#59046)

Signed-off-by: Martin Hickey <martin.hickey@ie.ibm.com>
This commit is contained in:
Martin Hickey
2026-10-02 03:17:56 +00:00
committed by GitHub
parent 8faacd707f
commit fcfacc3499
9 changed files with 760 additions and 163 deletions
+9 -3
View File
@@ -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()
+1 -1
View File
@@ -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"
@@ -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."""
@@ -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)
+32 -1
View File
@@ -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))
@@ -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
``<think>``). `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]
+191 -56
View File
@@ -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 </think>,
# <tool_call>, 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:"):
+2 -2
View File
@@ -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)
+6 -7
View File
@@ -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)