@tduncan / Nvidia-Optimization-Stack-VLLM / commits / 8b98b7d0b4

[Frontend] Add streaming parity tests and docs for derender (#57922)

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

Martin Hickey committed Sep 21, 2026 at 15:13 UTC 8b98b7d0b465920740cfa1acf5dde1b6a57d8d52

5 files changed +410 -101

docs/serving/online_serving/derenderer.md

+110 -1

@@ -6,7 +6,7 @@ This closes the loop for a token-in / token-out engine in disaggregated serving:

7 - **GPU less post processing**: Detokenization, reasoning parsing, and tool call parsing run on the same GPU less frontend that hosts `/render`

8 - **Parser parity**: The derenderer reuses vLLM's tool and reasoning parsers, so a disaggregated deployment produces the same `content`/`reasoning`/ `tool_calls` split as a standard `vllm serve` server

9 -- **Streaming and one-shot parsing**: Non-streaming calls send a complete `GenerateResponse` with all token IDs present and get one-shot parsing. Both endpoints also accept `stream: true`, taking one `GenerateStreamResponse` delta plus a client carried `stream_state` and returning `{chunk, stream_state}`. The chat endpoint's streaming path supports reasoning and tool call parsing while emitting the same `reasoning`/`content`/`tool_calls` deltas the generate streaming path would for the same token IDs

9 +- **Streaming and one-shot parsing**: Non-streaming calls send a complete `GenerateResponse` with all token IDs present and get one-shot parsing. Both endpoints also accept `stream: true`, taking one `GenerateStreamResponse` delta plus a client carried `stream_state` and returning `{chunk, stream_state}`. The chat endpoint's streaming path supports reasoning and tool call parsing while emitting the same `reasoning`/`content`/`tool_calls` deltas the generate streaming path would for the same token IDs (see [Streaming](#streaming))

11 Both endpoints are hosted by the GPU less rendering server started with [`vllm launch render`](../../cli/launch/render.md), alongside the `/render`

12 endpoints.

@@ -27,8 +27,10 @@ The derender step needs more than the engine's `token_ids`. It also consumes the

28 - Chat Completions Derender API (`/v1/chat/completions/derender`)

29 - Post process a single `GenerateResponse` into a `ChatCompletionResponse`

30 + - With `stream: true`, post process one `GenerateStreamResponse` chunk into a `ChatCompletionStreamResponse` chunk

31 - Completions Derender API (`/v1/completions/derender`)

32 - Post process a list of `GenerateResponse` objects (one per prompt) into a `CompletionResponse`

33 + - With `stream: true`, post process one `GenerateStreamResponse` chunk into a `CompletionStreamResponse` chunk

35 ## Request format

@@ -50,8 +52,54 @@ Each request wraps the engine's `GenerateResponse`(s) together with the caller m

52 --8<-- "vllm/entrypoints/scale_out/token_in_token_out/protocol.py:derender-completion-request"

53 ```

55 +Streaming requests set `stream: true` and carry one generate chunk (`generate_chunk`) plus `stream_state` instead of a complete response.

56 +

57 +`/v1/chat/completions/derender` with `stream: true`:

58 +

59 +??? code

60 +

61 + ```python

62 + --8<-- "vllm/entrypoints/scale_out/token_in_token_out/protocol.py:derender-chat-stream-request"

63 + ```

64 +

65 +`/v1/completions/derender` with `stream: true`:

66 +

67 +??? code

68 +

69 + ```python

70 + --8<-- "vllm/entrypoints/scale_out/token_in_token_out/protocol.py:derender-completion-stream-request"

71 + ```

72 +

73 +Both return `{"chunk": ..., "stream_state": ...}`. `chunk` is a `ChatCompletionStreamResponse` or `CompletionStreamResponse` and `stream_state` goes with the next call.

74 +

75 Oversized payloads are rejected with a `400` before any `tokenizer.decode()` or parser runs.

77 +## Parser configuration

78 +

79 +The derenderer builds its tool and reasoning parsers from its own server flags plus the `chat_request` sent with each call. It can't see how the request was rendered or served anywhere else, so for its output to match a `vllm serve` server:

80 +

81 +- Start the render server with the same `--tool-call-parser`, `--reasoning-parser`, `--enable-auto-tool-choice`, `--chat-template` and `--default-chat-template-kwargs` you'd give `vllm serve` for this model. If `/render` and `/derender` run on different servers, give both the same values.

82 +- Send the full `chat_request` that went to `/render`, not just `messages` and `tools`. Fields like `chat_template_kwargs`, `reasoning_effort`, `tool_choice` and `include_reasoning` change how the output is parsed.

83 +

84 +A mismatch doesn't fail. The parser just splits `reasoning`, `content` and `tool_calls` differently from what `vllm serve` would return.

85 +

86 +`/v1/completions/derender` only detokenizes. It never runs tool or reasoning parsers, the same as `/v1/completions` on `vllm serve`. It only reads `skip_special_tokens` from `completion_request`.

87 +

88 +## Streaming

89 +

90 +Streaming derender mirrors `/inference/v1/generate`. Set `stream: true` in the body on the same path and the endpoint takes one generate stream chunk instead of a complete response. It answers with JSON, not SSE. Each call turns one generate chunk into one derendered chunk and the client forwards that to its own caller.

91 +

92 +The server keeps no state between calls. Everything the next call needs is in `stream_state` which the client carries:

93 +

94 +- **Echo `stream_state` back.** Leave it out (or send `null`) on the first call, then send the `stream_state` from each response with the next request, unchanged. State that doesn't add up (e.g. `output_chunk_lens` that don't sum to the number of output tokens) is rejected with a `400`.

95 +- **One choice per chunk.** `/inference/v1/generate` sends one choice per SSE event and each derender call accepts at most one. More than one is rejected with a `400`.

96 +- **`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.

97 +- **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.

98 +- **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`.

99 +- **Don't forward `[DONE]`.** It marks the end of the generate stream and isn't a chunk.

100 +

101 +Streaming chunks don't carry logprobs yet, so `logprobs` on a generate chunk are dropped. The non-streaming endpoints do resolve them, including `token_id:N` placeholders.

102 +

103 ## Streaming cost

104

105 Streaming derender threads a client carried `stream_state` across per-chunk calls instead of keeping session state on the server.

@@ -118,3 +166,64 @@ print(response["choices"][0]["message"]["content"])

166 ```

167

168 Passing `chat_request` lets the derenderer run the configured tool and reasoning parsers. This means `response["choices"][0]["message"]` carries the same `content` / `reasoning` / `tool_calls` split a `vllm serve` server would produce. `chat_request` can only be omitted for a model with no tool or reasoning parser configured. A parser configured model rejects a missing `chat_request` with a 400 rather than silently falling back to plain detokenization.

169 +

170 +### Streaming example

171 +

172 +The same round trip, streamed. `/render` keeps `stream` and `stream_options` on the `GenerateRequest` it returns, so the generate call streams too. Each generate chunk goes through derender with the `stream_state` from the previous call (see [Streaming](#streaming)).

173 +

174 +```python

175 +import json

176 +

177 +import httpx

178 +

179 +MODEL = "meta-llama/Llama-3.2-1B-Instruct"

180 +RENDER = "http://localhost:8100" # vllm launch render ...

181 +ENGINE = "http://localhost:8200" # token-in / token-out engine

182 +

183 +chat_request = {

184 + "model": MODEL,

185 + "messages": [{"role": "user", "content": "What is 2+2?"}],

186 + "max_tokens": 32,

187 + "stream": True,

188 + "stream_options": {"include_usage": True},

189 +}

190 +

191 +with httpx.Client(timeout=60.0) as client:

192 + # 1. Render: request -> token IDs (GPU less)

193 + generate_request = client.post(

194 + f"{RENDER}/v1/chat/completions/render", json=chat_request

195 + ).json()

196 + prompt_token_ids = generate_request["token_ids"]

197 +

198 + # 2. Generate: stream token IDs (token-in / token-out engine)

199 + stream_state = None # one per choice index when n > 1

200 + with client.stream(

201 + "POST", f"{ENGINE}/inference/v1/generate", json=generate_request

202 + ) as generate_stream:

203 + for line in generate_stream.iter_lines():

204 + if not line.startswith("data: ") or line == "data: [DONE]":

205 + continue

206 +

207 + # 3. Derender: one generate chunk -> one chat completion chunk

208 + derendered = client.post(

209 + f"{RENDER}/v1/chat/completions/derender",

210 + json={

211 + "stream": True,

212 + "model": MODEL,

213 + "generate_chunk": json.loads(line[len("data: ") :]),

214 + "stream_state": stream_state,

215 + "prompt_tokens": len(prompt_token_ids),

216 + "prompt_token_ids": prompt_token_ids,

217 + "chat_request": chat_request,

218 + },

219 + ).json()

220 + stream_state = derendered["stream_state"]

221 +

222 + chunk = derendered["chunk"]

223 + for choice in chunk["choices"]:

224 + print(choice["delta"].get("content") or "", end="", flush=True)

225 + if chunk.get("usage"):

226 + print(f"\n{chunk['usage']}")

227 +```

228 +

229 +With a tool or reasoning parser configured, `delta` also carries `reasoning` and `tool_calls`, the same deltas `/v1/chat/completions` streams for those tokens.

tests/entrypoints/scale_out/derender/test_derender_parity.py

+181 -8

@@ -3,20 +3,25 @@

4 """Round trip parity CI: render -> generate -> derender -> render == render.

6 -Pins the coupled parsing path (``/v1/chat/completions`` on a normal GPU

6 +Pins the coupled parsing path (`/v1/chat/completions` on a normal GPU

7 server which parses generated tokens incrementally as they stream out)

8 -against the disaggregated path (``OnlineDerenderer.derender_chat`` via

9 -``/v1/chat/completions/derender`` which parses the same tokens all at once,

10 -out of process). A standard ``vllm serve`` GPU server mounts both, so one

8 +against the disaggregated path (`OnlineDerenderer.derender_chat` via

9 +`/v1/chat/completions/derender` which parses the same tokens all at once,

10 +out of process). A standard `vllm serve` GPU server mounts both, so one

11 real generation lets both parsers run on identical input.

13 Both paths consume the same generated token IDs (extracted from the coupled

14 -response's ``token_ids`` via ``return_token_ids=True``), so generation

14 +response's `token_ids` via `return_token_ids=True`), so generation

15 nondeterminism is irrelevant. The only variable under test is whether the

16 two parsing code paths agree. Parity is asserted unconditionally as only the

17 stronger per case assertions (e.g. "a tool call was produced") are gated

18 behind the marker actually having been emitted since a 1.5B model is not

19 -guaranteed to emit ``<think>`` / ``<tool_call>``.

19 +guaranteed to emit `<think>` / `<tool_call>`.

20 +

21 +The streaming cases do the same for `stream=true`. Each coupled chunk's

22 +`token_ids` becomes one `/inference/v1/generate` stream chunk, fed

23 +through the chunked derender path with `stream_state` threaded across calls

24 +the way a client would, so both parsers see the same chunk boundaries.

25 """

27 import json

@@ -24,7 +29,12 @@ import json

29 import httpx

30 import pytest

31 import pytest_asyncio

32 +import regex as re

34 +from tests.entrypoints.scale_out.derender.utils import (

35 + assemble_stream,

36 + stream_chat_derender,

37 +)

38 from tests.utils import RemoteOpenAIServer

40 MODEL = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"

@@ -228,8 +238,8 @@ async def test_parity_logprobs(client):

238 """token_id:N resolution parity vs. the coupled server's real strings.

239

240 A real disaggregated worker only has token IDs so it emits logprobs

231 - with ``token_id:N`` placeholders (``return_tokens_as_token_ids=True``

232 - reproduces that shape here). ``/derender`` must resolve those

241 + with `token_id:N` placeholders (`return_tokens_as_token_ids=True`

242 + reproduces that shape here). `/derender` must resolve those

243 placeholders to the same token strings/bytes the coupled server

244 resolves them to directly.

245 """

@@ -267,3 +277,166 @@ async def test_parity_logprobs(client):

277 for d_entry, r_entry in zip(d_content, r_content):

278 assert d_entry["token"] == r_entry["token"]

279 assert d_entry["bytes"] == r_entry["bytes"]

280 +

281 +

282 +# ---------------------------------------------------------------------------

283 +# Streaming helpers

284 +# ---------------------------------------------------------------------------

285 +

286 +

287 +async def _coupled_stream(

288 + client: httpx.AsyncClient, messages: list[dict], **extra

289 +) -> tuple[list[dict], list[int]]:

290 + """Stream `/v1/chat/completions` and return the choices that carry

291 + `token_ids` along with the prompt token IDs.

292 +

293 + The leading role only chunk has no `token_ids` and no counterpart on

294 + `/inference/v1/generate`, so it is dropped.

295 + """

296 + choices: list[dict] = []

297 + prompt_token_ids: list[int] | None = None

298 + async with client.stream(

299 + "POST",

300 + "/v1/chat/completions",

301 + json={

302 + "model": MODEL,

303 + "messages": messages,

304 + "temperature": 0,

305 + "max_tokens": 128,

306 + "return_token_ids": True,

307 + "stream": True,

308 + **extra,

309 + },

310 + ) as resp:

311 + assert resp.status_code == 200, await resp.aread()

312 + async for line in resp.aiter_lines():

313 + if not line.startswith("data: "):

314 + continue

315 + data = line[len("data: ") :]

316 + if data == "[DONE]":

317 + break

318 + chunk = json.loads(data)

319 + assert "error" not in chunk, chunk

320 + if chunk.get("prompt_token_ids") is not None:

321 + prompt_token_ids = chunk["prompt_token_ids"]

322 + choices.extend(

323 + ch for ch in chunk["choices"] if ch.get("token_ids") is not None

324 + )

325 + assert prompt_token_ids is not None

326 + return choices, prompt_token_ids

327 +

328 +

329 +def _tool_call_id_shape(tool_call_id: str | None) -> str | None:

330 + """Drop the random suffix so IDs compare by format, e.g. `chatcmpl-tool-`.

331 + Deterministic IDs such as kimi_k2's `functions.<name>:<idx>` are kept

332 + whole."""

333 + if tool_call_id is None:

334 + return None

335 + return re.sub(r"[0-9a-f]{16}$", "", tool_call_id)

336 +

337 +

338 +async def _run_stream_parity_case(

339 + client: httpx.AsyncClient,

340 + messages: list[dict],

341 + finish_only_tail: bool = False,

342 + **extra,

343 +) -> tuple[dict, list[dict]]:

344 + """Stream the coupled request, replay its chunks through the streaming

345 + derender endpoint and assert the assembled messages agree. Returns the

346 + assembled coupled message and the coupled choices.

347 +

348 + Derender gets the engine's finish reason, since `"tool_calls"` is a chat

349 + level rewrite of `"stop"` that derender has to reapply itself. With

350 + `finish_only_tail` it goes on a trailing chunk with no tokens.

351 + """

352 + coupled_choices, prompt_token_ids = await _coupled_stream(client, messages, **extra)

353 + finish_reason = coupled_choices[-1]["finish_reason"]

354 + disagg = await stream_chat_derender(

355 + client,

356 + [tid for ch in coupled_choices for tid in ch["token_ids"]],

357 + [len(ch["token_ids"]) for ch in coupled_choices]

358 + + ([0] if finish_only_tail else []),

359 + {"model": MODEL, "messages": messages, **extra},

360 + len(prompt_token_ids),

361 + prompt_token_ids,

362 + finish_reason="stop" if finish_reason == "tool_calls" else finish_reason,

363 + )

364 + coupled = assemble_stream(coupled_choices)

365 + for message in (coupled, disagg):

366 + for tc in message["tool_calls"]:

367 + tc["id"] = _tool_call_id_shape(tc["id"])

368 + assert disagg == coupled

369 + return coupled, coupled_choices

370 +

371 +

372 +# ---------------------------------------------------------------------------

373 +# Streaming parity cases

374 +# ---------------------------------------------------------------------------

375 +

376 +

377 +@pytest.mark.asyncio

378 +async def test_stream_parity_plain(client):

379 + """Plain streamed detokenization parity."""

380 + messages = [

381 + {"role": "user", "content": "What is 2+2? Answer in one short sentence."}

382 + ]

383 + await _run_stream_parity_case(client, messages)

384 +

385 +

386 +@pytest.mark.asyncio

387 +async def test_stream_parity_reasoning(client):

388 + """Streamed reasoning/content split parity."""

389 + messages = [{"role": "user", "content": "What is 17 times 23? Think it through."}]

390 + coupled, _ = await _run_stream_parity_case(

391 + client, messages, include_reasoning=True, max_tokens=256

392 + )

393 +

394 + if not coupled["reasoning"]:

395 + pytest.skip("Model did not emit a <think> block")

396 +

397 +

398 +@pytest.mark.asyncio

399 +async def test_stream_parity_tool_call(client):

400 + """Streamed tool call parity under `tool_choice="auto"`, where a tool

401 + call also rewrites `finish_reason` to `"tool_calls"`."""

402 + messages = [{"role": "user", "content": "What's the weather in Paris?"}]

403 + coupled, _ = await _run_stream_parity_case(

404 + client, messages, tools=TOOLS, tool_choice="auto", max_tokens=512

405 + )

406 +

407 + if not coupled["tool_calls"]:

408 + pytest.skip("Model did not emit a tool call")

409 +

410 +

411 +@pytest.mark.asyncio

412 +@pytest.mark.parametrize(

413 + "chunking", ["engine_steps", "multi_token", "finish_only_tail"]

414 +)

415 +async def test_stream_parity_reasoning_and_tool_call(client, chunking):

416 + """Streamed reasoning + forced tool call parity across chunk shapes.

417 +

418 + `multi_token` batches tokens per chunk with `stream_interval` the way

419 + speculative decoding does, so replay has to reuse the original chunk

420 + boundaries. `finish_only_tail` sends the finish reason on a chunk with no

421 + tokens, which must still flush buffered tool call arguments.

422 + """

423 + messages = [{"role": "user", "content": "What's the weather in Paris?"}]

424 + extra: dict = {

425 + "tools": TOOLS,

426 + "tool_choice": FORCE_WEATHER_TOOL,

427 + "include_reasoning": True,

428 + "max_tokens": 512,

429 + }

430 + if chunking == "multi_token":

431 + extra["stream_interval"] = 4

432 + coupled, coupled_choices = await _run_stream_parity_case(

433 + client,

434 + messages,

435 + finish_only_tail=chunking == "finish_only_tail",

436 + **extra,

437 + )

438 +

439 + if chunking == "multi_token":

440 + assert any(len(ch["token_ids"]) > 1 for ch in coupled_choices)

441 + if not (coupled["reasoning"] and coupled["tool_calls"]):

442 + pytest.skip("Model did not emit both a <think> block and a tool call")

tests/entrypoints/scale_out/derender/test_derender_stream.py

+12 -92

@@ -5,24 +5,24 @@

6 Tests are split into two layers:

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.

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 The parser path is covered both with a deterministic stub parser and

12 - with the real ``HarmonyParser`` (skipped without ``openai_harmony``).

12 + with the real `HarmonyParser` (skipped without `openai_harmony`).

14 2. Integration tests (require a running render server): covers the full

15 HTTP round-trip through the streaming endpoint. Marked with

16 - ``@pytest.mark.asyncio`` and gated by the ``server`` / ``client``

17 - fixtures from the sibling ``test_derender.py``.

16 + `@pytest.mark.asyncio` and gated by the `server` / `client`

17 + fixtures from the sibling `test_derender.py`.

18 """

20 import json

21 -from collections.abc import Callable

22 import pytest

23 import pytest_asyncio

25 +from tests.entrypoints.scale_out.derender.utils import stream_chat_derender

26 from vllm.entrypoints.generate.base.protocol import (

27 DeltaFunctionCall,

28 DeltaMessage,

@@ -356,7 +356,7 @@ class TestDetokenizeDelta:

356 """prev_tokens must not grow with the number of chunks (bounded transport).

357

358 Guards that the carried decode window is a small constant

359 - tail, so cumulative ``stream_state`` transport is O(n) and not O(n^2).

359 + tail, so cumulative `stream_state` transport is O(n) and not O(n^2).

360 """

361 token_ids = tokenizer.encode(

362 "a reasonably long ascii stream of tokens used to exercise the "

@@ -1616,86 +1616,6 @@ async def _render_parser_chat(client, messages: list[dict]) -> dict:

1616 return resp.json()

1617

1618

1619 -async def _stream_chat_derender(

1620 - client,

1621 - output_ids: list[int],

1622 - chunk_sizes: list[int],

1623 - chat_request: dict,

1624 - prompt_tokens: int,

1625 - prompt_token_ids: list[int],

1626 - on_chunk: Callable[[list[dict]], None] | None = None,

1627 -) -> dict:

1628 - """Feed `output_ids` through the streaming chat derender endpoint in

1629 - the given `chunk_sizes`, threading `stream_state` across calls and

1630 - return the assembled message.

1631 -

1632 - If `on_chunk` is given, it is called after every chunk with a

1633 - snapshot (deep copy) of the `tool_calls` accumulator so far, letting

1634 - callers assert properties of the intermediate deltas (e.g. monotonic

1635 - argument growth) rather than only the final assembled result.

1636 - """

1637 - state = None

1638 - content = ""

1639 - reasoning = ""

1640 - tool_calls: list[dict] = []

1641 - finish_reason = None

1642 -

1643 - pos = 0

1644 - for i, size in enumerate(chunk_sizes):

1645 - tids = output_ids[pos : pos + size]

1646 - pos += size

1647 - is_last = i == len(chunk_sizes) - 1

1648 - resp = await client.post(

1649 - "/v1/chat/completions/derender",

1650 - json={

1651 - "stream": True,

1652 - "model": PARSER_MODEL,

1653 - "generate_chunk": {

1654 - "request_id": "stream-test",

1655 - "choices": [

1656 - {

1657 - "index": 0,

1658 - "token_ids": tids,

1659 - "finish_reason": "stop" if is_last else None,

1660 - }

1661 - ],

1662 - },

1663 - "stream_state": state,

1664 - "prompt_tokens": prompt_tokens,

1665 - "prompt_token_ids": prompt_token_ids,

1666 - "chat_request": chat_request,

1667 - },

1668 - )

1669 - assert resp.status_code == 200, resp.text

1670 - data = resp.json()

1671 - state = data["stream_state"]

1672 - delta = data["chunk"]["choices"][0]["delta"]

1673 - content += delta.get("content") or ""

1674 - reasoning += delta.get("reasoning") or ""

1675 - for tc in delta.get("tool_calls") or []:

1676 - idx = tc["index"]

1677 - while len(tool_calls) <= idx:

1678 - tool_calls.append({"id": None, "name": None, "arguments": ""})

1679 - if tc.get("id"):

1680 - tool_calls[idx]["id"] = tc["id"]

1681 - fn = tc.get("function") or {}

1682 - if fn.get("name"):

1683 - tool_calls[idx]["name"] = fn["name"]

1684 - if fn.get("arguments"):

1685 - tool_calls[idx]["arguments"] += fn["arguments"]

1686 - if is_last:

1687 - finish_reason = data["chunk"]["choices"][0]["finish_reason"]

1688 - if on_chunk is not None:

1689 - on_chunk([dict(tc) for tc in tool_calls])

1690 -

1691 - return {

1692 - "content": content or None,

1693 - "reasoning": reasoning or None,

1694 - "tool_calls": tool_calls,

1695 - "finish_reason": finish_reason,

1696 - }

1697 -

1698 -

1619 @pytest.mark.asyncio

1620 @pytest.mark.parametrize(

1621 "chunking",

@@ -1755,7 +1675,7 @@ async def test_stream_parsed_matches_batch_reasoning(

1675 "triples": [3] * (n // 3) + [n % 3],

1676 }[chunking]

1677 chunk_sizes = [c for c in chunk_sizes if c > 0]

1758 - streamed = await _stream_chat_derender(

1678 + streamed = await stream_chat_derender(

1679 parser_client,

1680 output_ids,

1681 chunk_sizes,

@@ -1823,7 +1743,7 @@ async def test_stream_parsed_matches_batch_tool_call(parser_client, parser_token

1743 if tool_calls and tool_calls[0]["arguments"]:

1744 _snapshots.append(tool_calls[0]["arguments"])

1745

1826 - streamed = await _stream_chat_derender(

1746 + streamed = await stream_chat_derender(

1747 parser_client,

1748 output_ids,

1749 chunk_sizes,

@@ -1868,7 +1788,7 @@ async def test_stream_parsed_cjk_across_chunk_boundaries(

1788 "messages": messages,

1789 "include_reasoning": True,

1790 }

1871 - streamed = await _stream_chat_derender(

1791 + streamed = await stream_chat_derender(

1792 parser_client,

1793 output_ids,

1794 [1] * len(output_ids),

@@ -1896,7 +1816,7 @@ async def test_stream_parsed_include_reasoning_false(parser_client, parser_token

1816 "messages": messages,

1817 "include_reasoning": False,

1818 }

1899 - streamed = await _stream_chat_derender(

1819 + streamed = await stream_chat_derender(

1820 parser_client,

1821 output_ids,

1822 [1] * len(output_ids),

tests/entrypoints/scale_out/derender/utils.py new

+103

@@ -0,0 +1,103 @@

1 +# SPDX-License-Identifier: Apache-2.0

2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

3 +"""Helpers shared by the streaming chat derender tests."""

4 +

5 +from collections.abc import Callable

6 +

7 +import httpx

8 +

9 +

10 +def assemble_stream(choices: list[dict]) -> dict:

11 + """Fold streamed chat choices into one message.

12 +

13 + A tool call index must keep the first ID it was given. On the derender

14 + side that is what pinning the IDs in `stream_state` guarantees.

15 + """

16 + content = ""

17 + reasoning = ""

18 + tool_calls: list[dict] = []

19 + finish_reason = None

20 + for choice in choices:

21 + delta = choice["delta"]

22 + content += delta.get("content") or ""

23 + reasoning += delta.get("reasoning") or ""

24 + for tc in delta.get("tool_calls") or []:

25 + idx = tc["index"]

26 + while len(tool_calls) <= idx:

27 + tool_calls.append({"id": None, "name": None, "arguments": ""})

28 + if tc.get("id"):

29 + assert tool_calls[idx]["id"] in (None, tc["id"]), (

30 + f"tool call {idx} changed ID mid stream"

31 + )

32 + tool_calls[idx]["id"] = tc["id"]

33 + fn = tc.get("function") or {}

34 + if fn.get("name"):

35 + tool_calls[idx]["name"] = fn["name"]

36 + tool_calls[idx]["arguments"] += fn.get("arguments") or ""

37 + finish_reason = choice.get("finish_reason") or finish_reason

38 + return {

39 + "content": content or None,

40 + "reasoning": reasoning or None,

41 + "tool_calls": tool_calls,

42 + "finish_reason": finish_reason,

43 + }

44 +

45 +

46 +async def stream_chat_derender(

47 + client: httpx.AsyncClient,

48 + output_ids: list[int],

49 + chunk_sizes: list[int],

50 + chat_request: dict,

51 + prompt_tokens: int,

52 + prompt_token_ids: list[int],

53 + finish_reason: str = "stop",

54 + on_chunk: Callable[[list[dict]], None] | None = None,

55 +) -> dict:

56 + """Feed `output_ids` through the streaming chat derender endpoint in

57 + the given `chunk_sizes`, threading `stream_state` across calls and

58 + return the assembled message.

59 +

60 + `finish_reason` goes on the last chunk, so a trailing size of 0 sends it

61 + on a chunk with no tokens.

62 +

63 + If `on_chunk` is given, it is called after every chunk with the

64 + `tool_calls` assembled so far, letting callers assert properties of the

65 + intermediate deltas (e.g. monotonic argument growth) rather than only

66 + the final assembled result.

67 + """

68 + state = None

69 + choices: list[dict] = []

70 + pos = 0

71 + for i, size in enumerate(chunk_sizes):

72 + tids = output_ids[pos : pos + size]

73 + pos += size

74 + is_last = i == len(chunk_sizes) - 1

75 + resp = await client.post(

76 + "/v1/chat/completions/derender",

77 + json={

78 + "stream": True,

79 + "model": chat_request["model"],

80 + "generate_chunk": {

81 + "request_id": "stream-test",

82 + "choices": [

83 + {

84 + "index": 0,

85 + "token_ids": tids,

86 + "finish_reason": finish_reason if is_last else None,

87 + }

88 + ],

89 + },

90 + "stream_state": state,

91 + "prompt_tokens": prompt_tokens,

92 + "prompt_token_ids": prompt_token_ids,

93 + "chat_request": chat_request,

94 + },

95 + )

96 + assert resp.status_code == 200, resp.text

97 + data = resp.json()

98 + state = data["stream_state"]

99 + choices.extend(data["chunk"]["choices"])

100 + if on_chunk is not None:

101 + on_chunk(assemble_stream(choices)["tool_calls"])

102 +

103 + return assemble_stream(choices)

vllm/entrypoints/scale_out/token_in_token_out/protocol.py

@@ -602,6 +602,7 @@ class DerenderChatStreamRequest(BaseModel):

602 the client carried ``stream_state``.

603 """

604

605 + # --8<-- [start:derender-chat-stream-request]

606 stream: Literal[True]

607

608 model: str | None = None

@@ -629,6 +630,7 @@ class DerenderChatStreamRequest(BaseModel):

630

631 chat_request: ChatCompletionRequest | None = None

632 """The original (post adjust_request) ChatCompletionRequest from /render."""

633 + # --8<-- [end:derender-chat-stream-request]

634

635

636 class DerenderCompletionStreamRequest(BaseModel):

@@ -639,6 +641,7 @@ class DerenderCompletionStreamRequest(BaseModel):

641 returns the derendered chunk plus updated state.

642 """

643

644 + # --8<-- [start:derender-completion-stream-request]

645 stream: Literal[True]

646

647 model: str | None = None

@@ -653,6 +656,7 @@ class DerenderCompletionStreamRequest(BaseModel):

656

657 completion_request: CompletionRequest | None = None

658 """The original (post adjust_request) CompletionRequest from /render."""

659 + # --8<-- [end:derender-completion-stream-request]

660

661

662 class DerenderChatStreamResponse(BaseModel):