@tduncan / Nvidia-Optimization-Stack-VLLM / commits / 0a31372e5f

[Frontend] Add detokenization streaming derender for disaggregated serving (#47301)

Signed-off-by: Martin Hickey <martin.hickey@ie.ibm.com>

Martin Hickey committed Jul 30, 2026 at 02:01 UTC 0a31372e5fe7cc8bbaf96df0a3e88db9040cf8d5
6 files changed +1377 -32
tests/entrypoints/scale_out/derender/test_derender_stream.py new
+765
@@ -0,0 +1,765 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +
4 +"""Unit tests for streaming derender.
5 +
6 +Tests are split into two layers:
7 +
8 +1. Unit tests (no server): covers ``_detokenize_delta`` correctness
9 + (chunked == one-shot) and ``derender_completion_stream`` /
10 + ``derender_chat_stream`` logic via a real tokenizer on a tiny model.
11 +
12 +2. Integration tests (require a running render server): covers the full
13 + HTTP round-trip through the streaming endpoint. Marked with
14 + ``@pytest.mark.asyncio`` and gated by the ``server`` / ``client``
15 + fixtures from the sibling ``test_derender.py``.
16 +"""
17 +
18 +import pytest
19 +import pytest_asyncio
20 +
21 +from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
22 + DerenderStreamState,
23 + GenerateResponseStreamChoice,
24 + GenerateStreamResponse,
25 +)
26 +
27 +MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
28 +
29 +
30 +# ---------------------------------------------------------------------------
31 +# Helpers shared across tests
32 +# ---------------------------------------------------------------------------
33 +
34 +
35 +def _make_stream_chunk(
36 + token_ids: list[int],
37 + index: int = 0,
38 + finish_reason: str | None = None,
39 + request_id: str = "test-req",
40 + usage: dict | None = None,
41 +) -> GenerateStreamResponse:
42 + """Build a GenerateStreamResponse SSE chunk."""
43 + from vllm.entrypoints.openai.engine.protocol import UsageInfo
44 +
45 + return GenerateStreamResponse(
46 + request_id=request_id,
47 + choices=[
48 + GenerateResponseStreamChoice(
49 + index=index,
50 + token_ids=token_ids,
51 + finish_reason=finish_reason,
52 + )
53 + ],
54 + usage=UsageInfo(**usage) if usage else None,
55 + )
56 +
57 +
58 +def _make_usage_chunk(
59 + completion_tokens: int,
60 + prompt_tokens: int = 0,
61 + request_id: str = "test-req",
62 +) -> GenerateStreamResponse:
63 + """Build a usage only final SSE chunk (empty choices)."""
64 + from vllm.entrypoints.openai.engine.protocol import UsageInfo
65 +
66 + return GenerateStreamResponse(
67 + request_id=request_id,
68 + choices=[],
69 + usage=UsageInfo(
70 + prompt_tokens=prompt_tokens,
71 + completion_tokens=completion_tokens,
72 + total_tokens=prompt_tokens + completion_tokens,
73 + ),
74 + )
75 +
76 +
77 +# ---------------------------------------------------------------------------
78 +# Unit tests — no running server
79 +# ---------------------------------------------------------------------------
80 +
81 +
82 +@pytest.fixture(scope="module")
83 +def tokenizer():
84 + """Load the tiny tokenizer used across unit tests."""
85 + from vllm.tokenizers import get_tokenizer
86 +
87 + return get_tokenizer(MODEL_NAME)
88 +
89 +
90 +@pytest.fixture(scope="module")
91 +def derenderer(tokenizer):
92 + """Construct a minimal OnlineDerenderer backed by a stub renderer."""
93 + from unittest.mock import MagicMock
94 +
95 + from vllm.renderers.online_derenderer import OnlineDerenderer
96 +
97 + renderer = MagicMock()
98 + renderer.get_tokenizer.return_value = tokenizer
99 +
100 + model_config = MagicMock()
101 + model_config.hf_config.model_type = "llama"
102 + model_config.model = MODEL_NAME
103 +
104 + return OnlineDerenderer(
105 + model_config=model_config,
106 + renderer=renderer,
107 + request_logger=None,
108 + chat_template=None,
109 + chat_template_content_format="string",
110 + trust_request_chat_template=False,
111 + enable_auto_tools=False,
112 + tool_parser=None,
113 + reasoning_parser=None,
114 + )
115 +
116 +
117 +class TestDetokenizeDelta:
118 + """_detokenize_delta: chunked decode must equal one shot decode."""
119 +
120 + def _one_shot(self, tokenizer, token_ids: list[int]) -> str:
121 + return tokenizer.decode(token_ids, skip_special_tokens=True)
122 +
123 + def _chunked(self, derenderer, tokenizer, chunks: list[list[int]]) -> str:
124 + state = DerenderStreamState()
125 + parts: list[str] = []
126 + for delta in chunks:
127 + text, state = derenderer._detokenize_delta(
128 + tokenizer, delta, state, skip_special_tokens=True
129 + )
130 + parts.append(text)
131 + return "".join(parts)
132 +
133 + def test_single_chunk(self, derenderer, tokenizer):
134 + """All tokens in one chunk == one shot decode."""
135 + token_ids = tokenizer.encode("Hello world")[:8]
136 + assert self._chunked(derenderer, tokenizer, [token_ids]) == self._one_shot(
137 + tokenizer, token_ids
138 + )
139 +
140 + def test_two_equal_chunks(self, derenderer, tokenizer):
141 + """Split in half and reassemble == one shot."""
142 + token_ids = tokenizer.encode("Hello world from streaming derender")[:12]
143 + mid = len(token_ids) // 2
144 + chunks = [token_ids[:mid], token_ids[mid:]]
145 + assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
146 + tokenizer, token_ids
147 + )
148 +
149 + def test_single_token_per_chunk(self, derenderer, tokenizer):
150 + """One token per chunk (most granular streaming) == one shot."""
151 + token_ids = tokenizer.encode("incremental detokenization test")[:10]
152 + chunks = [[t] for t in token_ids]
153 + assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
154 + tokenizer, token_ids
155 + )
156 +
157 + def test_empty_delta_passthrough(self, derenderer, tokenizer):
158 + """An empty delta (usage only chunk) emits empty string and preserves state."""
159 + token_ids = tokenizer.encode("Hello")[:4]
160 + _, state = derenderer._detokenize_delta(
161 + tokenizer, token_ids, DerenderStreamState(), skip_special_tokens=True
162 + )
163 + text, new_state = derenderer._detokenize_delta(
164 + tokenizer, [], state, skip_special_tokens=True
165 + )
166 + assert text == ""
167 + assert new_state.prev_tokens == state.prev_tokens
168 + assert new_state.prefix_offset == state.prefix_offset
169 + assert new_state.read_offset == state.read_offset
170 +
171 + def test_multibyte_char_split_across_chunks(self, derenderer, tokenizer):
172 + """A CJK/emoji char straddling chunk boundaries == one shot.
173 +
174 + Regression test for held back trailing incomplete UTF-8 byte
175 + sequences being dropped when the rebuild window marks them as
176 + already read (see #46159).
177 + """
178 + token_ids = tokenizer.encode("Hello ✅ world 日本語")[:16]
179 + chunks = [[t] for t in token_ids]
180 + assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
181 + tokenizer, token_ids
182 + )
183 +
184 + def test_state_carries_across_calls(self, derenderer, tokenizer):
185 + """Decode state threads across calls. Text still matches one shot."""
186 + t1 = tokenizer.encode("Hello")[:2]
187 + t2 = tokenizer.encode(" world")[:2]
188 + state = DerenderStreamState()
189 + text1, state = derenderer._detokenize_delta(tokenizer, t1, state)
190 + text2, state = derenderer._detokenize_delta(tokenizer, t2, state)
191 + assert text1 + text2 == self._one_shot(tokenizer, t1 + t2)
192 + # Offsets are rebased to the carried tail each chunk
193 + assert state.prefix_offset == 0
194 +
195 + def test_state_window_stays_bounded(self, derenderer, tokenizer):
196 + """prev_tokens must not grow with the number of chunks (bounded transport).
197 +
198 + Guards that the carried decode window is a small constant
199 + tail, so cumulative ``stream_state`` transport is O(n) and not O(n^2).
200 + """
201 + token_ids = tokenizer.encode(
202 + "a reasonably long ascii stream of tokens used to exercise the "
203 + "window bound across many single token chunks so the carried "
204 + "state cannot grow linearly with the generation length"
205 + )
206 + assert len(token_ids) > 32
207 + state = DerenderStreamState()
208 + max_window = 0
209 + for tok in token_ids:
210 + _, state = derenderer._detokenize_delta(tokenizer, [tok], state)
211 + max_window = max(max_window, len(state.prev_tokens))
212 + # Bounded by a small constant independent of len(token_ids)
213 + assert max_window <= 32
214 +
215 + def test_n_independent_streams_same_result(self, derenderer, tokenizer):
216 + """N parallel streams with the same token sequence give the same text."""
217 + token_ids = tokenizer.encode("parallel streams")[:8]
218 + mid = len(token_ids) // 2
219 +
220 + results = []
221 + for _ in range(3):
222 + state = DerenderStreamState()
223 + text, state = derenderer._detokenize_delta(
224 + tokenizer, token_ids[:mid], state
225 + )
226 + text2, _ = derenderer._detokenize_delta(tokenizer, token_ids[mid:], state)
227 + results.append(text + text2)
228 +
229 + assert len(set(results)) == 1, "All independent streams must produce same text"
230 + assert results[0] == self._one_shot(tokenizer, token_ids)
231 +
232 +
233 +class TestDerenderCompletionStream:
234 + """derender_completion_stream: streaming output parity with one shot."""
235 +
236 + @pytest.mark.asyncio
237 + async def test_chunked_equals_oneshot(self, derenderer, tokenizer):
238 + """Sum of streaming text chunks == one shot tokenizer.decode."""
239 + token_ids = tokenizer.encode("streaming completion test")[:10]
240 + mid = len(token_ids) // 2
241 +
242 + state = DerenderStreamState()
243 + chunk1, state = await derenderer.derender_completion_stream(
244 + model=MODEL_NAME,
245 + generate_chunk=_make_stream_chunk(token_ids[:mid]),
246 + state=state,
247 + )
248 + chunk2, _ = await derenderer.derender_completion_stream(
249 + model=MODEL_NAME,
250 + generate_chunk=_make_stream_chunk(token_ids[mid:], finish_reason="stop"),
251 + state=state,
252 + )
253 +
254 + streamed_text = chunk1.choices[0].text + chunk2.choices[0].text
255 + one_shot = tokenizer.decode(token_ids, skip_special_tokens=True)
256 + assert streamed_text == one_shot
257 +
258 + @pytest.mark.asyncio
259 + async def test_usage_chunk_passthrough(self, derenderer, tokenizer):
260 + """Usage only final chunk (empty choices) is passed through correctly."""
261 + usage_chunk = _make_usage_chunk(completion_tokens=10, prompt_tokens=5)
262 + chunk, state = await derenderer.derender_completion_stream(
263 + model=MODEL_NAME,
264 + generate_chunk=usage_chunk,
265 + )
266 + assert chunk.choices == []
267 + assert chunk.usage is not None
268 + assert chunk.usage.completion_tokens == 10
269 + assert chunk.usage.prompt_tokens == 5
270 +
271 + @pytest.mark.asyncio
272 + async def test_prompt_tokens_in_usage(self, derenderer, tokenizer):
273 + """prompt_tokens is correctly forwarded into usage on a usage chunk."""
274 + token_ids = tokenizer.encode("hello")[:3]
275 + usage_chunk = _make_usage_chunk(
276 + completion_tokens=len(token_ids), prompt_tokens=7
277 + )
278 + chunk, _ = await derenderer.derender_completion_stream(
279 + model=MODEL_NAME,
280 + generate_chunk=usage_chunk,
281 + prompt_tokens=7,
282 + )
283 + assert chunk.usage is not None
284 + assert chunk.usage.prompt_tokens == 7
285 +
286 + @pytest.mark.asyncio
287 + async def test_none_state_initialises_correctly(self, derenderer, tokenizer):
288 + """Passing state=None (first call) initialises an empty DerenderStreamState."""
289 + token_ids = tokenizer.encode("hello")[:4]
290 + chunk, state = await derenderer.derender_completion_stream(
291 + model=MODEL_NAME,
292 + generate_chunk=_make_stream_chunk(token_ids),
293 + state=None,
294 + )
295 + assert isinstance(state, DerenderStreamState)
296 + assert chunk.choices[0].text == tokenizer.decode(
297 + token_ids, skip_special_tokens=True
298 + )
299 +
300 + @pytest.mark.asyncio
301 + async def test_skip_special_tokens_threaded(self, derenderer, tokenizer):
302 + """completion_request.skip_special_tokens is honored (not hardcoded True)."""
303 + from vllm.entrypoints.openai.completion.protocol import CompletionRequest
304 +
305 + eos = tokenizer.eos_token_id
306 + if eos is None:
307 + pytest.skip("tokenizer has no eos token to exercise special stripping")
308 + token_ids = tokenizer.encode("hi")[:2] + [eos]
309 +
310 + async def _text(skip: bool) -> str:
311 + req = CompletionRequest(
312 + model=MODEL_NAME, prompt="x", skip_special_tokens=skip
313 + )
314 + chunk, _ = await derenderer.derender_completion_stream(
315 + model=MODEL_NAME,
316 + generate_chunk=_make_stream_chunk(token_ids),
317 + completion_request=req,
318 + )
319 + return chunk.choices[0].text
320 +
321 + # skip=False must retain the special token; skip=True must strip it.
322 + assert await _text(False) != await _text(True)
323 +
324 + @pytest.mark.asyncio
325 + async def test_finish_reason_forwarded(self, derenderer, tokenizer):
326 + """finish_reason from the generate chunk reaches the derendered choice."""
327 + token_ids = tokenizer.encode("done")[:2]
328 + chunk, _ = await derenderer.derender_completion_stream(
329 + model=MODEL_NAME,
330 + generate_chunk=_make_stream_chunk(token_ids, finish_reason="length"),
331 + )
332 + assert chunk.choices[0].finish_reason == "length"
333 +
334 +
335 +class TestDerenderChatStream:
336 + """derender_chat_stream: plain detok branch (no parser)."""
337 +
338 + @pytest.mark.asyncio
339 + async def test_role_on_first_chunk_only(self, derenderer, tokenizer):
340 + """role='assistant' appears in the first chunk, not subsequent ones."""
341 + token_ids = tokenizer.encode("hello world")[:6]
342 + mid = len(token_ids) // 2
343 +
344 + state = DerenderStreamState()
345 + chunk1, state = await derenderer.derender_chat_stream(
346 + model=MODEL_NAME,
347 + generate_chunk=_make_stream_chunk(token_ids[:mid]),
348 + state=state,
349 + )
350 + chunk2, _ = await derenderer.derender_chat_stream(
351 + model=MODEL_NAME,
352 + generate_chunk=_make_stream_chunk(token_ids[mid:], finish_reason="stop"),
353 + state=state,
354 + )
355 +
356 + assert chunk1.choices[0].delta.role == "assistant"
357 + assert chunk2.choices[0].delta.role is None
358 +
359 + @pytest.mark.asyncio
360 + async def test_chunked_equals_oneshot(self, derenderer, tokenizer):
361 + """Sum of streaming content deltas == one shot decode."""
362 + token_ids = tokenizer.encode("streaming chat derender text")[:10]
363 + mid = len(token_ids) // 2
364 +
365 + state = DerenderStreamState()
366 + chunk1, state = await derenderer.derender_chat_stream(
367 + model=MODEL_NAME,
368 + generate_chunk=_make_stream_chunk(token_ids[:mid]),
369 + state=state,
370 + )
371 + chunk2, _ = await derenderer.derender_chat_stream(
372 + model=MODEL_NAME,
373 + generate_chunk=_make_stream_chunk(token_ids[mid:]),
374 + state=state,
375 + )
376 +
377 + streamed = (chunk1.choices[0].delta.content or "") + (
378 + chunk2.choices[0].delta.content or ""
379 + )
380 + one_shot = tokenizer.decode(token_ids, skip_special_tokens=True)
381 + assert streamed == one_shot
382 +
383 + @pytest.mark.asyncio
384 + @pytest.mark.parametrize("with_chat_request", [True, False])
385 + async def test_parser_raises_not_implemented(self, tokenizer, with_chat_request):
386 + """Stream chat with a parser active must fail closed (NotImplementedError).
387 +
388 + Checks that a parser configured model must 501 even when ``chat_request`` is
389 + omitted, otherwise reasoning/tool markup would leak into ``delta.content`
390 + via the plain detok fallback.
391 + """
392 + from unittest.mock import MagicMock
393 +
394 + from vllm.renderers.online_derenderer import OnlineDerenderer
395 +
396 + renderer = MagicMock()
397 + renderer.get_tokenizer.return_value = tokenizer
398 +
399 + model_config = MagicMock()
400 + model_config.hf_config.model_type = "llama"
401 + model_config.model = MODEL_NAME
402 +
403 + # Construct a derenderer WITH a tool/reasoning parser active.
404 + # ParserManager.get_parser returns None when no parser is named, so
405 + # we inject a mock parser class directly.
406 + dr = OnlineDerenderer(
407 + model_config=model_config,
408 + renderer=renderer,
409 + request_logger=None,
410 + chat_template=None,
411 + chat_template_content_format="string",
412 + )
413 + dr.parser = MagicMock() # simulate a parser being active
414 +
415 + chat_request = MagicMock() if with_chat_request else None
416 + token_ids = tokenizer.encode("hello")[:3]
417 +
418 + with pytest.raises(NotImplementedError):
419 + await dr.derender_chat_stream(
420 + model=MODEL_NAME,
421 + generate_chunk=_make_stream_chunk(token_ids),
422 + state=None,
423 + chat_request=chat_request,
424 + )
425 +
426 +
427 +class TestDerenderStreamStateValidation:
428 + """DerenderStreamState rejects malformed caller supplied offsets/lengths."""
429 +
430 + def test_negative_prefix_offset_rejected(self):
431 + from pydantic import ValidationError
432 +
433 + with pytest.raises(ValidationError):
434 + DerenderStreamState(prefix_offset=-1)
435 +
436 + def test_negative_read_offset_rejected(self):
437 + from pydantic import ValidationError
438 +
439 + with pytest.raises(ValidationError):
440 + DerenderStreamState(read_offset=-1)
441 +
442 + def test_prev_tokens_over_cap_rejected(self):
443 + from pydantic import ValidationError
444 +
445 + with pytest.raises(ValidationError):
446 + DerenderStreamState(prev_tokens=["a"] * 1025)
447 +
448 + def test_prev_tokens_at_cap_accepted(self):
449 + state = DerenderStreamState(prev_tokens=["a"] * 1024)
450 + assert len(state.prev_tokens) == 1024
451 +
452 +
453 +class TestServingDerenderStreamErrorHandling:
454 + """Malformed stream_state must surface as 400 and not an unhandled 500."""
455 +
456 + def _make_serving(self, side_effect: Exception):
457 + from unittest.mock import AsyncMock, MagicMock
458 +
459 + from vllm.entrypoints.scale_out.derender.serving import ServingDerender
460 +
461 + models = MagicMock()
462 + models.is_base_model.return_value = True
463 + models.model_config = MagicMock()
464 +
465 + online_derenderer = MagicMock()
466 + online_derenderer.derender_completion_stream = AsyncMock(
467 + side_effect=side_effect
468 + )
469 + online_derenderer.derender_chat_stream = AsyncMock(side_effect=side_effect)
470 +
471 + return ServingDerender(models=models, online_derenderer=online_derenderer)
472 +
473 + @pytest.mark.asyncio
474 + @pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
475 + async def test_completion_stream_bad_state_returns_400(self, exc):
476 + from vllm.entrypoints.openai.engine.protocol import ErrorResponse
477 + from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
478 + DerenderCompletionStreamRequest,
479 + )
480 +
481 + serving = self._make_serving(exc)
482 + request = DerenderCompletionStreamRequest(
483 + stream=True,
484 + model=MODEL_NAME,
485 + generate_chunk=_make_stream_chunk([1, 2]),
486 + stream_state=DerenderStreamState(),
487 + )
488 + result = await serving.derender_completion_stream_response(request)
489 + assert isinstance(result, ErrorResponse)
490 + assert result.error.code == 400
491 +
492 + @pytest.mark.asyncio
493 + @pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
494 + async def test_chat_stream_bad_state_returns_400(self, exc):
495 + from vllm.entrypoints.openai.engine.protocol import ErrorResponse
496 + from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
497 + DerenderChatStreamRequest,
498 + )
499 +
500 + serving = self._make_serving(exc)
501 + request = DerenderChatStreamRequest(
502 + stream=True,
503 + model=MODEL_NAME,
504 + generate_chunk=_make_stream_chunk([1, 2]),
505 + stream_state=DerenderStreamState(),
506 + )
507 + result = await serving.derender_chat_stream_response(request)
508 + assert isinstance(result, ErrorResponse)
509 + assert result.error.code == 400
510 +
511 +
512 +# ---------------------------------------------------------------------------
513 +# Integration tests — require a live render server
514 +# ---------------------------------------------------------------------------
515 +
516 +
517 +@pytest.fixture(scope="module")
518 +def server():
519 + from tests.utils import RemoteLaunchRenderServer
520 +
521 + with RemoteLaunchRenderServer(MODEL_NAME, []) as remote_server:
522 + yield remote_server
523 +
524 +
525 +@pytest_asyncio.fixture
526 +async def client(server):
527 + import httpx
528 +
529 + async with httpx.AsyncClient(
530 + base_url=server.url_for(""), timeout=30.0
531 + ) as http_client:
532 + yield http_client
533 +
534 +
535 +async def _render_chat(client) -> dict:
536 + """Render a minimal chat request and return the GenerateRequest dict."""
537 +
538 + resp = await client.post(
539 + "/v1/chat/completions/render",
540 + json={
541 + "model": MODEL_NAME,
542 + "messages": [{"role": "user", "content": "Hello"}],
543 + },
544 + )
545 + assert resp.status_code == 200
546 + return resp.json()
547 +
548 +
549 +@pytest.mark.asyncio
550 +async def test_streaming_completion_derender_roundtrip(client):
551 + """Streaming completions derender: chunked text == non streaming text."""
552 + gen_req = await _render_chat(client)
553 + token_ids: list[int] = gen_req["token_ids"][:8]
554 + mid = len(token_ids) // 2
555 + chunk1_ids, chunk2_ids = token_ids[:mid], token_ids[mid:]
556 +
557 + # Non streaming baseline.
558 + non_stream_resp = await client.post(
559 + "/v1/completions/derender",
560 + json={
561 + "model": MODEL_NAME,
562 + "generate_responses": [
563 + {
564 + "request_id": "test-ns",
565 + "choices": [
566 + {
567 + "index": 0,
568 + "token_ids": token_ids,
569 + "finish_reason": "stop",
570 + }
571 + ],
572 + }
573 + ],
574 + },
575 + )
576 + assert non_stream_resp.status_code == 200
577 + expected_text = non_stream_resp.json()["choices"][0]["text"]
578 +
579 + # Streaming call 1.
580 + r1 = await client.post(
581 + "/v1/completions/derender",
582 + json={
583 + "stream": True,
584 + "model": MODEL_NAME,
585 + "generate_chunk": {
586 + "request_id": "test-s",
587 + "choices": [
588 + {"index": 0, "token_ids": chunk1_ids, "finish_reason": None}
589 + ],
590 + },
591 + "stream_state": None,
592 + },
593 + )
594 + assert r1.status_code == 200
595 + d1 = r1.json()
596 + text1 = d1["chunk"]["choices"][0]["text"]
597 + state1 = d1["stream_state"]
598 +
599 + # Streaming call 2 (final chunk).
600 + r2 = await client.post(
601 + "/v1/completions/derender",
602 + json={
603 + "stream": True,
604 + "model": MODEL_NAME,
605 + "generate_chunk": {
606 + "request_id": "test-s",
607 + "choices": [
608 + {"index": 0, "token_ids": chunk2_ids, "finish_reason": "stop"}
609 + ],
610 + },
611 + "stream_state": state1,
612 + },
613 + )
614 + assert r2.status_code == 200
615 + text2 = r2.json()["chunk"]["choices"][0]["text"]
616 +
617 + assert text1 + text2 == expected_text
618 +
619 +
620 +@pytest.mark.asyncio
621 +async def test_streaming_chat_derender_roundtrip(client):
622 + """Streaming chat derender (plain detok): chunked text == non streaming text."""
623 + gen_req = await _render_chat(client)
624 + token_ids: list[int] = gen_req["token_ids"][:8]
625 + mid = len(token_ids) // 2
626 + chunk1_ids, chunk2_ids = token_ids[:mid], token_ids[mid:]
627 +
628 + # Non streaming baseline.
629 + ns = await client.post(
630 + "/v1/chat/completions/derender",
631 + json={
632 + "model": MODEL_NAME,
633 + "generate_response": {
634 + "request_id": "test-ns",
635 + "choices": [
636 + {
637 + "index": 0,
638 + "token_ids": token_ids,
639 + "finish_reason": "stop",
640 + }
641 + ],
642 + },
643 + },
644 + )
645 + assert ns.status_code == 200
646 + expected_content = ns.json()["choices"][0]["message"]["content"]
647 +
648 + # Streaming call 1.
649 + r1 = await client.post(
650 + "/v1/chat/completions/derender",
651 + json={
652 + "stream": True,
653 + "model": MODEL_NAME,
654 + "generate_chunk": {
655 + "request_id": "test-s",
656 + "choices": [
657 + {"index": 0, "token_ids": chunk1_ids, "finish_reason": None}
658 + ],
659 + },
660 + "stream_state": None,
661 + },
662 + )
663 + assert r1.status_code == 200
664 + d1 = r1.json()
665 + text1 = d1["chunk"]["choices"][0]["delta"].get("content") or ""
666 + state1 = d1["stream_state"]
667 + # role=assistant on the first chunk
668 + assert d1["chunk"]["choices"][0]["delta"].get("role") == "assistant"
669 +
670 + # Streaming call 2.
671 + r2 = await client.post(
672 + "/v1/chat/completions/derender",
673 + json={
674 + "stream": True,
675 + "model": MODEL_NAME,
676 + "generate_chunk": {
677 + "request_id": "test-s",
678 + "choices": [
679 + {"index": 0, "token_ids": chunk2_ids, "finish_reason": "stop"}
680 + ],
681 + },
682 + "stream_state": state1,
683 + },
684 + )
685 + assert r2.status_code == 200
686 + d2 = r2.json()
687 + text2 = d2["chunk"]["choices"][0]["delta"].get("content") or ""
688 + # role must NOT be repeated on subsequent chunks
689 + assert d2["chunk"]["choices"][0]["delta"].get("role") is None
690 +
691 + assert text1 + text2 == expected_content
692 +
693 +
694 +@pytest.mark.asyncio
695 +async def test_streaming_derender_invalid_body_returns_400(client):
696 + """Missing required field in streaming request returns 400."""
697 + r = await client.post(
698 + "/v1/completions/derender",
699 + json={
700 + "stream": True,
701 + # missing required 'model' and 'generate_chunk'
702 + },
703 + )
704 + assert r.status_code == 400
705 +
706 +
707 +@pytest.mark.asyncio
708 +async def test_streaming_derender_non_object_body_returns_400(client):
709 + """A non object JSON body (e.g. a list) returns 400, not a 500."""
710 + r = await client.post(
711 + "/v1/completions/derender",
712 + json=[1, 2, 3],
713 + )
714 + assert r.status_code == 400
715 +
716 +
717 +@pytest.mark.asyncio
718 +async def test_streaming_usage_chunk(client):
719 + """Usage only final chunk is forwarded with correct token counts."""
720 + gen_req = await _render_chat(client)
721 + token_ids: list[int] = gen_req["token_ids"][:6]
722 + state: dict = {}
723 +
724 + # Send content chunk first.
725 + r1 = await client.post(
726 + "/v1/completions/derender",
727 + json={
728 + "stream": True,
729 + "model": MODEL_NAME,
730 + "generate_chunk": {
731 + "request_id": "usage-test",
732 + "choices": [
733 + {"index": 0, "token_ids": token_ids, "finish_reason": "stop"}
734 + ],
735 + },
736 + "stream_state": None,
737 + },
738 + )
739 + assert r1.status_code == 200
740 + state = r1.json()["stream_state"]
741 +
742 + # Send usage only final chunk.
743 + r2 = await client.post(
744 + "/v1/completions/derender",
745 + json={
746 + "stream": True,
747 + "model": MODEL_NAME,
748 + "generate_chunk": {
749 + "request_id": "usage-test",
750 + "choices": [],
751 + "usage": {
752 + "prompt_tokens": 10,
753 + "completion_tokens": len(token_ids),
754 + "total_tokens": 10 + len(token_ids),
755 + },
756 + },
757 + "stream_state": state,
758 + "prompt_tokens": 10,
759 + },
760 + )
761 + assert r2.status_code == 200
762 + d2 = r2.json()
763 + assert d2["chunk"]["choices"] == []
764 + assert d2["chunk"]["usage"]["prompt_tokens"] == 10
765 + assert d2["chunk"]["usage"]["completion_tokens"] == len(token_ids)
vllm/entrypoints/scale_out/derender/api_router.py
+64 -12
@@ -5,15 +5,17 @@ from http import HTTPStatus
5 from fastapi import APIRouter, Depends, Request
6 from fastapi.responses import JSONResponse
7
8 -from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionResponse
9 -from vllm.entrypoints.openai.completion.protocol import CompletionResponse
8 from vllm.entrypoints.openai.engine.protocol import ErrorResponse
9 from vllm.entrypoints.serve.utils.api_utils import validate_json_request
10 from vllm.logger import init_logger
11
12 from ..token_in_token_out.protocol import (
15 - DerenderChatRequest,
16 - DerenderCompletionRequest,
13 + DerenderChatRequestUnion,
14 + DerenderChatStreamRequest,
15 + DerenderChatStreamResponse,
16 + DerenderCompletionRequestUnion,
17 + DerenderCompletionStreamRequest,
18 + DerenderCompletionStreamResponse,
19 )
20 from .serving import ServingDerender
21
@@ -29,46 +31,96 @@ def derender(request: Request) -> ServingDerender | None:
31 @router.post(
32 "/v1/chat/completions/derender",
33 dependencies=[Depends(validate_json_request)],
32 - response_model=ChatCompletionResponse,
34 responses={
35 HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
36 HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
37 HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
38 },
39 )
39 -async def derender_chat_completion(request: DerenderChatRequest, raw_request: Request):
40 +async def derender_chat_completion(
41 + request: DerenderChatRequestUnion,
42 + raw_request: Request,
43 +):
44 + """Derender a generate response into a ChatCompletionResponse.
45 +
46 + Accepts both non-streaming (``stream=false``, default) and streaming
47 + (``stream=true``) request bodies on the same path; FastAPI validates and
48 + routes on the ``stream`` discriminator.
49 +
50 + Non-streaming: body is ``DerenderChatRequest`` (``generate_response`` with
51 + the complete token list). Returns a ``ChatCompletionResponse``.
52 +
53 + Streaming: body is ``DerenderChatStreamRequest`` (one ``generate_chunk``
54 + delta + optional ``stream_state``). Returns a ``DerenderChatStreamResponse``
55 + (``chunk`` + ``stream_state``). The client carries ``stream_state`` between
56 + successive calls, one per SSE chunk from ``/inference/v1/generate``.
57 + """
58 handler = derender(raw_request)
59 if handler is None:
60 raise NotImplementedError(
61 "The model does not support Chat Completions Derender API"
62 )
63
46 - result = await handler.derender_chat_response(request)
64 + if isinstance(request, DerenderChatStreamRequest):
65 + stream_result = await handler.derender_chat_stream_response(request)
66 + if isinstance(stream_result, ErrorResponse):
67 + return JSONResponse(
68 + content=stream_result.model_dump(),
69 + status_code=stream_result.error.code,
70 + )
71 + chunk, stream_state = stream_result
72 + response = DerenderChatStreamResponse(chunk=chunk, stream_state=stream_state)
73 + return JSONResponse(content=response.model_dump())
74
75 + result = await handler.derender_chat_response(request)
76 if isinstance(result, ErrorResponse):
77 return JSONResponse(content=result.model_dump(), status_code=result.error.code)
50 -
78 return JSONResponse(content=result.model_dump())
79
80
81 @router.post(
82 "/v1/completions/derender",
83 dependencies=[Depends(validate_json_request)],
57 - response_model=CompletionResponse,
84 responses={
85 HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
86 HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
87 HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
88 },
89 )
64 -async def derender_completion(request: DerenderCompletionRequest, raw_request: Request):
90 +async def derender_completion(
91 + request: DerenderCompletionRequestUnion,
92 + raw_request: Request,
93 +):
94 + """Derender a generate response into a CompletionResponse.
95 +
96 + Accepts both non-streaming (``stream=false``, default) and streaming
97 + (``stream=true``) request bodies on the same path.
98 +
99 + Non-streaming: body is ``DerenderCompletionRequest``. Returns a
100 + ``CompletionResponse``.
101 +
102 + Streaming: body is ``DerenderCompletionStreamRequest`` (one
103 + ``generate_chunk`` + optional ``stream_state``). Returns a
104 + ``DerenderCompletionStreamResponse`` (``chunk`` + ``stream_state``).
105 + """
106 handler = derender(raw_request)
107 if handler is None:
108 raise NotImplementedError("The model does not support Completions Derender API")
109
69 - result = await handler.derender_completion_response(request)
110 + if isinstance(request, DerenderCompletionStreamRequest):
111 + stream_result = await handler.derender_completion_stream_response(request)
112 + if isinstance(stream_result, ErrorResponse):
113 + return JSONResponse(
114 + content=stream_result.model_dump(),
115 + status_code=stream_result.error.code,
116 + )
117 + chunk, stream_state = stream_result
118 + response = DerenderCompletionStreamResponse(
119 + chunk=chunk, stream_state=stream_state
120 + )
121 + return JSONResponse(content=response.model_dump())
122
123 + result = await handler.derender_completion_response(request)
124 if isinstance(result, ErrorResponse):
125 return JSONResponse(content=result.model_dump(), status_code=result.error.code)
73 -
126 return JSONResponse(content=result.model_dump())
vllm/entrypoints/scale_out/derender/serving.py
+98 -3
@@ -4,8 +4,14 @@ import time
4 from typing import cast
5
6 import vllm.envs as envs
7 -from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionResponse
8 -from vllm.entrypoints.openai.completion.protocol import CompletionResponse
7 +from vllm.entrypoints.openai.chat_completion.protocol import (
8 + ChatCompletionResponse,
9 + ChatCompletionStreamResponse,
10 +)
11 +from vllm.entrypoints.openai.completion.protocol import (
12 + CompletionResponse,
13 + CompletionStreamResponse,
14 +)
15 from vllm.entrypoints.openai.engine.protocol import (
16 ErrorResponse,
17 UsageInfo,
@@ -28,7 +34,10 @@ from vllm.renderers.online_derenderer import OnlineDerenderer
34 from ..token_in_token_out.mm_serde import encode_mm_kwargs_item
35 from ..token_in_token_out.protocol import (
36 DerenderChatRequest,
37 + DerenderChatStreamRequest,
38 DerenderCompletionRequest,
39 + DerenderCompletionStreamRequest,
40 + DerenderStreamState,
41 GenerateResponse,
42 MultiModalFeatures,
43 PlaceholderRangeInfo,
@@ -200,7 +209,9 @@ class ServingDerender(BaseServing):
209 total_prompt_tokens,
210 total_completion_tokens,
211 ) = await self.online_derenderer.derender_completion(
203 - request.generate_responses, request.prompt_tokens
212 + request.generate_responses,
213 + request.prompt_tokens,
214 + completion_request=request.completion_request,
215 )
216
217 first = request.generate_responses[0]
@@ -237,6 +248,90 @@ class ServingDerender(BaseServing):
248 kv_transfer_params=kv_params,
249 )
250
251 + async def derender_chat_stream_response(
252 + self,
253 + request: DerenderChatStreamRequest,
254 + ) -> tuple[ChatCompletionStreamResponse, DerenderStreamState] | ErrorResponse:
255 + """Streaming counterpart to ``derender_chat_response``.
256 +
257 + Processes one ``GenerateStreamResponse`` chunk and returns the
258 + derendered chunk together with the updated client carried state.
259 +
260 + ``parser is None`` or no ``chat_request`` until reasoning/tool call
261 + functionality added in future PR.
262 + """
263 + error_check_ret = await self._check_model(request)
264 + if error_check_ret is not None:
265 + return error_check_ret
266 +
267 + try:
268 + chunk, updated_state = await self.online_derenderer.derender_chat_stream(
269 + model=request.model,
270 + generate_chunk=request.generate_chunk,
271 + state=request.stream_state,
272 + chat_request=request.chat_request,
273 + prompt_tokens=request.prompt_tokens,
274 + )
275 + except NotImplementedError as exc:
276 + return self.create_error_response(exc)
277 + except ValueError as exc:
278 + return self.create_error_response(str(exc))
279 + except (KeyError, IndexError) as exc:
280 + return self.create_error_response(
281 + f"invalid stream_state: detokenization failed ({exc!r})"
282 + )
283 +
284 + logger.debug(
285 + "derender_chat_stream request_id=%s model=%s delta_tokens=%d",
286 + request.generate_chunk.request_id,
287 + request.model,
288 + sum(
289 + len(c.token_ids) for c in request.generate_chunk.choices if c.token_ids
290 + ),
291 + )
292 + return chunk, updated_state
293 +
294 + async def derender_completion_stream_response(
295 + self,
296 + request: DerenderCompletionStreamRequest,
297 + ) -> tuple[CompletionStreamResponse, DerenderStreamState] | ErrorResponse:
298 + """Streaming counterpart to ``derender_completion_response``.
299 +
300 + Processes one ``GenerateStreamResponse`` chunk (one output sequence's
301 + delta) and returns the derendered chunk and updated state.
302 + """
303 + error_check_ret = await self._check_model(request)
304 + if error_check_ret is not None:
305 + return error_check_ret
306 +
307 + try:
308 + (
309 + chunk,
310 + updated_state,
311 + ) = await self.online_derenderer.derender_completion_stream(
312 + model=request.model,
313 + generate_chunk=request.generate_chunk,
314 + state=request.stream_state,
315 + prompt_tokens=request.prompt_tokens,
316 + completion_request=request.completion_request,
317 + )
318 + except ValueError as exc:
319 + return self.create_error_response(str(exc))
320 + except (KeyError, IndexError) as exc:
321 + return self.create_error_response(
322 + f"invalid stream_state: detokenization failed ({exc!r})"
323 + )
324 +
325 + logger.debug(
326 + "derender_completion_stream request_id=%s model=%s delta_tokens=%d",
327 + request.generate_chunk.request_id,
328 + request.model,
329 + sum(
330 + len(c.token_ids) for c in request.generate_chunk.choices if c.token_ids
331 + ),
332 + )
333 + return chunk, updated_state
334 +
335 @staticmethod
336 def _extract_mm_features(
337 engine_input: EngineInput,
vllm/entrypoints/scale_out/token_in_token_out/protocol.py
+165 -11
@@ -1,6 +1,6 @@
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 -from typing import Any
3 +from typing import Any, Literal, TypeAlias
4
5 from pydantic import (
6 BaseModel,
@@ -14,8 +14,12 @@ from vllm.config import ModelConfig
14 from vllm.entrypoints.openai.chat_completion.protocol import (
15 ChatCompletionLogProbs,
16 ChatCompletionRequest,
17 + ChatCompletionStreamResponse,
18 +)
19 +from vllm.entrypoints.openai.completion.protocol import (
20 + CompletionRequest,
21 + CompletionStreamResponse,
22 )
18 -from vllm.entrypoints.openai.completion.protocol import CompletionRequest
23 from vllm.entrypoints.openai.engine.protocol import StreamOptions, UsageInfo
24 from vllm.logprobs import Logprob
25 from vllm.renderers import TokenizeParams
@@ -252,21 +256,16 @@ class GenerateResponse(BaseModel):
256 )
257
258
255 -####### Derender (postprocessing) #######
256 -
257 -
259 class DerenderChatRequest(BaseModel):
260 """Request for the /v1/chat/completions/derender endpoint (non-streaming).
261
261 - Wraps a complete GenerateResponse and caller-supplied metadata needed to
262 - produce a fully-formed ChatCompletionResponse without a GPU.
263 -
264 - Streaming derender would require a separate endpoint design with
265 - incremental token delivery, ``OutputProcessor``-based detokenization,
266 - and ``parser.parse_delta()`` instead of ``parser.parse()``.
262 + Wraps a complete GenerateResponse and caller supplied metadata needed to
263 + produce a fully formed ChatCompletionResponse without a GPU.
264 """
265
266 # --8<-- [start:derender-chat-request]
267 + stream: Literal[False] = False
268 +
269 model: str
270 """Served model name."""
271
@@ -299,6 +298,8 @@ class DerenderCompletionRequest(BaseModel):
298 """
299
300 # --8<-- [start:derender-completion-request]
301 + stream: Literal[False] = False
302 +
303 model: str
304 """Served model name."""
305
@@ -330,3 +331,156 @@ class DerenderCompletionRequest(BaseModel):
331 f"generate_responses length ({len(self.generate_responses)})"
332 )
333 return self
334 +
335 +
336 +class DerenderStreamState(BaseModel):
337 + """Per sequence state for stateless streaming derender.
338 +
339 + The client carries this between successive per chunk HTTP calls to the
340 + streaming derender endpoint. All fields are plain JSON serializable data.
341 + No opaque tokenizer or parser internals are stored here.
342 +
343 + The detokenization strategy carries the incremental decode offsets
344 + directly rather than re-sending the whole token history each chunk.
345 + ``detokenize_incrementally`` only ever reads the trailing token window
346 + ``prev_tokens[prefix_offset:]``, so we carry just that tail plus the two
347 + offsets. Each chunk resumes exactly where the last one stopped, including
348 + any partially processed multi-byte character (tracked by ``read_offset``),
349 + then trims and rebases the window so it never grows with generation length.
350 +
351 + Performance:
352 + - Compute per chunk is O(delta). One ``detokenize_incrementally`` call per
353 + new token, independent of how many tokens preceded it.
354 + - Transport per chunk is O(window). The carried tail is bounded by the
355 + incremental detokenization offset, so cumulative bytes over the wire are
356 + O(n) rather than the O(n^2) a full history round trip would incur.
357 + """
358 +
359 + prev_tokens: list[str] = Field(default_factory=list)
360 + """Trailing decode window. Token strings from ``prefix_offset`` onward.
361 +
362 + Bounded, trimmed and rebased each chunk to the tail
363 + ``detokenize_incrementally`` still reads, so it does not grow with the
364 + number of chunks.
365 + """
366 +
367 + prefix_offset: int = Field(default=0, ge=0)
368 + """Prefix offset into ``prev_tokens`` for incremental detokenization."""
369 +
370 + read_offset: int = Field(default=0, ge=0)
371 + """Read offset into ``prev_tokens`` for incremental detokenization."""
372 +
373 + @field_validator("prev_tokens")
374 + @classmethod
375 + def _bound_prev_tokens(cls, v: list[str]) -> list[str]:
376 + # INITIAL_INCREMENTAL_DETOKENIZATION_OFFSET is small (5) and the trimmed
377 + # window is O(offset). A generous limit rejects unusually large or malformed
378 + # payloads without restricting legitimate multi-byte sequences.
379 + limit = 1024
380 + if len(v) > limit:
381 + raise ValueError(f"prev_tokens length ({len(v)}) exceeds maximum ({limit})")
382 + return v
383 +
384 + role_sent: bool = False
385 + """True once the initial ``role: "assistant"`` delta has been emitted.
386 +
387 + Prevents re-emitting the role on subsequent chunks even when the detok
388 + window is transiently empty (e.g. usage only final chunk).
389 + """
390 +
391 + # TODO: Properties used in follow on PR for tool call parsing
392 + last_content: str | None = None
393 + """Last emitted cumulative assistant content text."""
394 +
395 + last_reasoning: str | None = None
396 + """Last emitted cumulative reasoning text."""
397 +
398 + last_tool_call_ids: list[str] = Field(default_factory=list)
399 + """Stable tool-call IDs, assigned once when each call first appears.
400 +
401 + Prevents ID regeneration across re-parsing.
402 + """
403 +
404 +
405 +class DerenderChatStreamRequest(BaseModel):
406 + """One chunk streaming derender request for /v1/chat/completions/derender.
407 +
408 + The client sends one request per SSE chunk received from
409 + ``/inference/v1/generate``. Each request carries the generate chunk
410 + plus the ``stream_state`` returned by the previous call (``None`` on the
411 + first call). The response contains the derendered chunk and the updated
412 + state to be passed to the next call.
413 +
414 + This implements stateless no server side session. All mutable state lives in
415 + the client carried ``stream_state``.
416 + """
417 +
418 + stream: Literal[True]
419 +
420 + model: str
421 + generate_chunk: GenerateStreamResponse
422 + """One SSE chunk from ``/inference/v1/generate`` (``stream=True``)."""
423 +
424 + stream_state: DerenderStreamState | None = None
425 + """Client carried detok state from the previous call. ``None`` on first."""
426 +
427 + prompt_tokens: int | None = None
428 + """Prompt token count for usage. Forwarded from the render step."""
429 +
430 + chat_request: ChatCompletionRequest | None = None
431 + """The original (post adjust_request) ChatCompletionRequest from /render."""
432 +
433 +
434 +class DerenderCompletionStreamRequest(BaseModel):
435 + """One chunk streaming derender request for /v1/completions/derender.
436 +
437 + Parallel to ``DerenderChatStreamRequest`` for the completions endpoint.
438 + Each call processes one SSE chunk (one output sequence's delta) and
439 + returns the derendered chunk plus updated state.
440 + """
441 +
442 + stream: Literal[True]
443 +
444 + model: str
445 + generate_chunk: GenerateStreamResponse
446 + """One SSE chunk from ``/inference/v1/generate``."""
447 +
448 + stream_state: DerenderStreamState | None = None
449 + """Client-carried detok state. ``None`` on the first call."""
450 +
451 + prompt_tokens: int | None = None
452 + """Prompt token count for usage."""
453 +
454 + completion_request: CompletionRequest | None = None
455 + """The original (post adjust_request) CompletionRequest from /render."""
456 +
457 +
458 +class DerenderChatStreamResponse(BaseModel):
459 + """Response for one streaming chat derender chunk.
460 +
461 + Pairs the derendered SSE chunk with the updated client carried state to
462 + pass to the next call.
463 + """
464 +
465 + chunk: ChatCompletionStreamResponse
466 + stream_state: DerenderStreamState
467 +
468 +
469 +class DerenderCompletionStreamResponse(BaseModel):
470 + """Response for one streaming completions derender chunk.
471 +
472 + Parallel to ``DerenderChatStreamResponse`` for the completions endpoint.
473 + """
474 +
475 + chunk: CompletionStreamResponse
476 + stream_state: DerenderStreamState
477 +
478 +
479 +# Determines the type by checking the ``stream`` field's literal value. A body without
480 +# ``stream`` validates as the non-streaming member
481 +# (``stream`` defaults to ``False`` there), so FastAPI can validate and dispatch both
482 +# shapes on a single path.
483 +DerenderChatRequestUnion: TypeAlias = DerenderChatRequest | DerenderChatStreamRequest
484 +DerenderCompletionRequestUnion: TypeAlias = (
485 + DerenderCompletionRequest | DerenderCompletionStreamRequest
486 +)
vllm/entrypoints/serve/engine/typing.py
+4
@@ -17,7 +17,9 @@ from vllm.entrypoints.openai.completion.protocol import (
17 from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
18 from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
19 DerenderChatRequest,
20 + DerenderChatStreamRequest,
21 DerenderCompletionRequest,
22 + DerenderCompletionStreamRequest,
23 GenerateRequest,
24 GenerateResponse,
25 )
@@ -54,6 +56,7 @@ CompletionLikeRequest: TypeAlias = (
56 | TokenizeCompletionRequest
57 | DetokenizeRequest
58 | DerenderCompletionRequest
59 + | DerenderCompletionStreamRequest
60 )
61
62 ChatLikeRequest: TypeAlias = (
@@ -61,6 +64,7 @@ ChatLikeRequest: TypeAlias = (
64 | BatchChatCompletionRequest
65 | TokenizeChatRequest
66 | DerenderChatRequest
67 + | DerenderChatStreamRequest
68 )
69
70 SpeechToTextRequest: TypeAlias = TranscriptionRequest | TranslationRequest
vllm/renderers/online_derenderer.py
+281 -6
@@ -10,19 +10,29 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
10 ChatCompletionNamedToolChoiceParam,
11 ChatCompletionRequest,
12 ChatCompletionResponseChoice,
13 + ChatCompletionResponseStreamChoice,
14 + ChatCompletionStreamResponse,
15 ChatMessage,
16 )
17 from vllm.entrypoints.openai.completion.protocol import (
18 CompletionLogProbs,
19 + CompletionRequest,
20 CompletionResponseChoice,
21 + CompletionResponseStreamChoice,
22 + CompletionStreamResponse,
23 +)
24 +from vllm.entrypoints.openai.engine.protocol import DeltaMessage, ToolCall, UsageInfo
25 +from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
26 + DerenderStreamState,
27 + GenerateResponse,
28 + GenerateStreamResponse,
29 )
19 -from vllm.entrypoints.openai.engine.protocol import ToolCall
20 -from vllm.entrypoints.scale_out.token_in_token_out.protocol import GenerateResponse
30 from vllm.entrypoints.serve.utils.request_logger import RequestLogger
31 from vllm.logger import init_logger
32 from vllm.parser import Parser, ParserManager
33 from vllm.renderers import BaseRenderer
34 from vllm.tokenizers import TokenizerLike
35 +from vllm.tokenizers.detokenizer_utils import detokenize_incrementally
36 from vllm.utils import random_uuid
37 from vllm.utils.async_utils import make_async
38
@@ -168,9 +178,15 @@ class OnlineDerenderer:
178 tool_calls=tc_items,
179 )
180 else:
171 - # No parser: plain detokenization.
181 + # No parser: plain detokenization honouring the request's
182 + # skip_special_tokens (default True when no request was given).
183 + skip_special = (
184 + chat_request.skip_special_tokens
185 + if chat_request is not None
186 + else True
187 + )
188 decoded_text = tokenizer.decode(
173 - choice.token_ids, skip_special_tokens=True
189 + choice.token_ids, skip_special_tokens=skip_special
190 )
191 message = ChatMessage(role="assistant", content=decoded_text)
192
@@ -185,23 +201,200 @@ class OnlineDerenderer:
201
202 return choices
203
204 + def _detokenize_delta(
205 + self,
206 + tokenizer: TokenizerLike,
207 + delta_token_ids: list[int],
208 + state: DerenderStreamState,
209 + skip_special_tokens: bool = True,
210 + spaces_between_special_tokens: bool = True,
211 + ) -> tuple[str, DerenderStreamState]:
212 + """Incrementally detokenize ``delta_token_ids`` from prior stream state.
213 +
214 + Resumes decoding from the offsets carried in ``state`` rather than
215 + replaying token history. ``state.prev_tokens`` holds the trailing decode
216 + window (from ``prefix_offset`` onward) that ``detokenize_incrementally``
217 + still needs to reproduce any partially read multi-byte character
218 + (tracked by ``read_offset``). The delta tokens are fed straight onto it.
219 +
220 + The window is bounded. ``detokenize_incrementally`` never reads before
221 + ``prefix_offset``, so after processing we trim ``prev_tokens`` to that
222 + tail and rebase the offsets to it. State transport therefore stays
223 + O(window) per chunk instead of re-sending the full token history.
224 +
225 + Args:
226 + tokenizer: The tokenizer to decode with.
227 + delta_token_ids: New token IDs from this generate chunk.
228 + state: Client carried detok state from the previous call.
229 + skip_special_tokens: Passed through to the tokenizer.
230 + spaces_between_special_tokens: Passed through to the tokenizer.
231 +
232 + Returns:
233 + (new_text, updated_state) — the delta text for this chunk and the
234 + state to pass to the next call.
235 + """
236 + prev_tokens = list(state.prev_tokens)
237 + prefix_offset = state.prefix_offset
238 + read_offset = state.read_offset
239 +
240 + text_parts: list[str] = []
241 + for tok_id in delta_token_ids:
242 + # prev_tokens is a (possibly empty) list, never None, so this
243 + # always takes the non first iter path and only consumes
244 + # all_input_ids[-1].
245 + new_toks, text, prefix_offset, read_offset = detokenize_incrementally(
246 + tokenizer=tokenizer,
247 + all_input_ids=[tok_id],
248 + prev_tokens=prev_tokens,
249 + prefix_offset=prefix_offset,
250 + read_offset=read_offset,
251 + skip_special_tokens=skip_special_tokens,
252 + spaces_between_special_tokens=spaces_between_special_tokens,
253 + )
254 + prev_tokens = prev_tokens + new_toks
255 + text_parts.append(text)
256 +
257 + # Trim to the tail still readable by detokenize_incrementally
258 + # (everything before prefix_offset is dead) and rebase the offsets so
259 + # the carried window stays bounded regardless of generation length.
260 + trimmed = prev_tokens[prefix_offset:]
261 + updated_state = state.model_copy(
262 + update={
263 + "prev_tokens": trimmed,
264 + "prefix_offset": 0,
265 + "read_offset": read_offset - prefix_offset,
266 + }
267 + )
268 + return "".join(text_parts), updated_state
269 +
270 + async def derender_chat_stream(
271 + self,
272 + model: str,
273 + generate_chunk: GenerateStreamResponse,
274 + state: DerenderStreamState | None = None,
275 + chat_request: ChatCompletionRequest | None = None,
276 + prompt_tokens: int | None = None,
277 + ) -> tuple[ChatCompletionStreamResponse, DerenderStreamState]:
278 + """Process one GenerateStreamResponse chunk for streaming chat derender.
279 +
280 + TODO: parse path for reasoning and tool calls is implemented in future PR.
281 +
282 + Unlike OpenAI's API, which always emits ``role: "assistant"`` on the
283 + very first chunk, this emits it on the first chunk with a non empty
284 + ``choices`` list. A leading usage only chunk therefore defers the
285 + role to the following content chunk instead of sending an empty
286 + role only delta.
287 +
288 + Args:
289 + model: Model name for the response object.
290 + generate_chunk: One SSE chunk from ``/inference/v1/generate``.
291 + state: Client carried detok state (``None`` for first call).
292 + chat_request: Original ChatCompletionRequest from ``/render``.
293 + prompt_tokens: Prompt token count for the usage chunk.
294 +
295 + Returns:
296 + (chunk, updated_state) — the derendered SSE chunk and the state
297 + the client must pass to the next call.
298 + """
299 + if state is None:
300 + state = DerenderStreamState()
301 +
302 + if self.parser is not None:
303 + # TODO: Follow on PR will implement the parse path. Check on the
304 + # parser alone (fail closed). A parser configured model must never
305 + # fall through to plain detok on the streaming path, even when
306 + # ``chat_request`` is omitted or reasoning/tool markup would leak
307 + # into ``delta.content``.
308 + raise NotImplementedError(
309 + "Streaming chat derender is not yet supported for models with "
310 + "a reasoning or tool parser configured. Use the non-streaming "
311 + "derender endpoint (stream=false) for parsed output."
312 + )
313 +
314 + # A single DerenderStreamState is threaded through every choice in
315 + # this chunk. Correct only when there is at most one choice per SSE
316 + # event (n=1, one call per index), as the streaming derender
317 + # protocol assumes. Multiple choices sharing one chunk would corrupt
318 + # each other's detok window.
319 + if len(generate_chunk.choices) > 1:
320 + raise ValueError(
321 + "derender_chat_stream expects at most one choice per chunk"
322 + )
323 +
324 + tokenizer = self.renderer.get_tokenizer()
325 + skip_special = (
326 + chat_request.skip_special_tokens if chat_request is not None else True
327 + )
328 + stream_choices: list[ChatCompletionResponseStreamChoice] = []
329 + updated_state = state
330 +
331 + for choice in generate_chunk.choices:
332 + delta_tids = choice.token_ids or []
333 + new_text, updated_state = self._detokenize_delta(
334 + tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special
335 + )
336 +
337 + include_role = not updated_state.role_sent
338 + if include_role:
339 + updated_state = updated_state.model_copy(update={"role_sent": True})
340 +
341 + delta = DeltaMessage(
342 + role="assistant" if include_role else None,
343 + content=new_text if new_text else None,
344 + )
345 + stream_choices.append(
346 + ChatCompletionResponseStreamChoice(
347 + index=choice.index,
348 + delta=delta,
349 + finish_reason=choice.finish_reason,
350 + )
351 + )
352 +
353 + usage: UsageInfo | None = None
354 + if generate_chunk.usage is not None:
355 + u = generate_chunk.usage
356 + pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
357 + ct = u.completion_tokens or 0
358 + usage = UsageInfo(
359 + prompt_tokens=pt,
360 + completion_tokens=ct,
361 + total_tokens=pt + ct,
362 + )
363 +
364 + chunk = ChatCompletionStreamResponse(
365 + id=generate_chunk.request_id,
366 + model=model,
367 + choices=stream_choices,
368 + usage=usage,
369 + )
370 + return chunk, updated_state
371 +
372 async def derender_completion(
373 self,
374 generate_responses: list[GenerateResponse],
375 prompt_tokens: list[int] | None = None,
376 + completion_request: CompletionRequest | None = None,
377 ) -> tuple[list[CompletionResponseChoice], int, int]:
193 - return await self._derender_completion_async(generate_responses, prompt_tokens)
378 + return await self._derender_completion_async(
379 + generate_responses, prompt_tokens, completion_request
380 + )
381
382 def _derender_completion(
383 self,
384 generate_responses: list[GenerateResponse],
385 prompt_tokens: list[int] | None = None,
386 + completion_request: CompletionRequest | None = None,
387 ) -> tuple[list[CompletionResponseChoice], int, int]:
388 n = len(generate_responses)
389 prompt_tokens_list: list[int] = (
390 prompt_tokens if prompt_tokens is not None else [0] * n
391 )
392
393 + skip_special = (
394 + completion_request.skip_special_tokens
395 + if completion_request is not None
396 + else True
397 + )
398 tokenizer = self.renderer.get_tokenizer()
399 choices: list[CompletionResponseChoice] = []
400 total_prompt_tokens = 0
@@ -217,7 +410,7 @@ class OnlineDerenderer:
410 )
411
412 decoded_text = tokenizer.decode(
220 - choice.token_ids, skip_special_tokens=True
413 + choice.token_ids, skip_special_tokens=skip_special
414 )
415 completion_logprobs = None
416 if choice.logprobs is not None:
@@ -239,6 +432,88 @@ class OnlineDerenderer:
432
433 return choices, total_prompt_tokens, total_completion_tokens
434
435 + async def derender_completion_stream(
436 + self,
437 + model: str,
438 + generate_chunk: GenerateStreamResponse,
439 + state: DerenderStreamState | None = None,
440 + prompt_tokens: int | None = None,
441 + completion_request: CompletionRequest | None = None,
442 + ) -> tuple[CompletionStreamResponse, DerenderStreamState]:
443 + """Process one GenerateStreamResponse chunk for streaming completions.
444 +
445 + Each call takes one SSE chunk from ``/inference/v1/generate`` plus the
446 + client carried ``stream_state`` and returns a ``CompletionStreamResponse``
447 + chunk and the updated state.
448 +
449 + The generate stream emits one choice per SSE event, so this method
450 + processes one output sequence at a time. For ``n > 1`` the client
451 + maintains one ``DerenderStreamState`` per ``choice.index``.
452 +
453 + Args:
454 + model: Model name for the response object.
455 + generate_chunk: One SSE chunk from ``/inference/v1/generate``.
456 + state: Client carried detok state (``None`` → first call).
457 + prompt_tokens: Prompt token count for usage (from the render step).
458 + completion_request: Original CompletionRequest from ``/render``;
459 + supplies ``skip_special_tokens``.
460 +
461 + Returns:
462 + (chunk, updated_state) — the derendered chunk and updated state.
463 + """
464 + if state is None:
465 + state = DerenderStreamState()
466 +
467 + # See the equivalent check in derender_chat_stream: a single
468 + # DerenderStreamState is threaded through every choice in this
469 + # chunk, so more than one choice per chunk would corrupt the
470 + # detok window across choices.
471 + if len(generate_chunk.choices) > 1:
472 + raise ValueError(
473 + "derender_completion_stream expects at most one choice per chunk"
474 + )
475 +
476 + tokenizer = self.renderer.get_tokenizer()
477 + skip_special = (
478 + completion_request.skip_special_tokens
479 + if completion_request is not None
480 + else True
481 + )
482 + stream_choices: list[CompletionResponseStreamChoice] = []
483 + updated_state = state
484 +
485 + for choice in generate_chunk.choices:
486 + delta_tids = choice.token_ids or []
487 + new_text, updated_state = self._detokenize_delta(
488 + tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special
489 + )
490 + stream_choices.append(
491 + CompletionResponseStreamChoice(
492 + index=choice.index,
493 + text=new_text,
494 + finish_reason=choice.finish_reason,
495 + )
496 + )
497 +
498 + usage: UsageInfo | None = None
499 + if generate_chunk.usage is not None:
500 + u = generate_chunk.usage
501 + pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
502 + ct = u.completion_tokens or 0
503 + usage = UsageInfo(
504 + prompt_tokens=pt,
505 + completion_tokens=ct,
506 + total_tokens=pt + ct,
507 + )
508 +
509 + chunk = CompletionStreamResponse(
510 + id=generate_chunk.request_id,
511 + model=model,
512 + choices=stream_choices,
513 + usage=usage,
514 + )
515 + return chunk, updated_state
516 +
517
518 def _parse_token_id_placeholder(token: str) -> int | None:
519 """Extract token ID from a 'token_id:N' placeholder string."""