[Frontend] expose stream_interval as req sampling param (#49754)
Signed-off-by: walterbm <walter.beller.morales@gmail.com> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Walter Beller-Morales committed
Jul 26, 2026 at 23:30 UTC
8040ef242662af2f07614d78862c53338f046878
5 files changed
+114
tests/v1/engine/test_output_processor.py
+76
@@ -141,6 +141,82 @@ def test_incremental_detokenization(
141
assert not output_processor.has_unfinished_requests()
142
143
144
+def test_request_stream_interval_raises_but_not_below_engine_default(
145
+ dummy_test_vectors,
146
+):
147
+ """A per-request stream_interval can raise the interval above the engine
148
+ default but not below it (values under the default clamp up), without
149
+ altering the generated text."""
150
+ engine_stream_interval = 5
151
+ # Request 0 (below the default) clamps up to 5; request 1 raises it to 10.
152
+ request_stream_intervals = [1, 10]
153
+ output_processor = OutputProcessor(
154
+ dummy_test_vectors.tokenizer,
155
+ log_stats=False,
156
+ stream_interval=engine_stream_interval,
157
+ )
158
+
159
+ requests = [
160
+ EngineCoreRequest(
161
+ request_id=f"request-{idx}-int",
162
+ external_req_id=f"request-{idx}",
163
+ prompt_token_ids=prompt_tokens,
164
+ mm_features=None,
165
+ arrival_time=0,
166
+ lora_request=None,
167
+ cache_salt=None,
168
+ data_parallel_rank=None,
169
+ sampling_params=SamplingParams(
170
+ skip_special_tokens=False,
171
+ spaces_between_special_tokens=False,
172
+ output_kind=RequestOutputKind.DELTA,
173
+ stop=[],
174
+ include_stop_str_in_output=False,
175
+ stream_interval=request_stream_intervals[idx],
176
+ ),
177
+ pooling_params=None,
178
+ )
179
+ for idx, prompt_tokens in enumerate(
180
+ dummy_test_vectors.prompt_tokens[: len(request_stream_intervals)]
181
+ )
182
+ ]
183
+
184
+ num_requests = len(requests)
185
+ engine_core = MockEngineCore(
186
+ tokens_list=dummy_test_vectors.generation_tokens[:num_requests],
187
+ prompts_list=dummy_test_vectors.prompt_tokens[:num_requests],
188
+ request_ids=[req.request_id for req in requests],
189
+ )
190
+
191
+ for request, prompt in zip(requests, dummy_test_vectors.prompt_strings):
192
+ output_processor.add_request(request, prompt)
193
+
194
+ gen_strings: dict[str, str] = {}
195
+ gen_tokens: dict[str, list[int]] = {}
196
+ while outputs := engine_core.get_outputs():
197
+ for request_output in output_processor.process_outputs(outputs).request_outputs:
198
+ request_id = request_output.request_id
199
+ new_tokens = request_output.outputs[0].token_ids
200
+ if request_id not in gen_strings:
201
+ gen_strings[request_id] = request_output.outputs[0].text
202
+ gen_tokens[request_id] = list(new_tokens)
203
+ assert len(new_tokens) == 1, f"{len(new_tokens)=}"
204
+ continue
205
+ gen_strings[request_id] += request_output.outputs[0].text
206
+ gen_tokens[request_id].extend(new_tokens)
207
+ if not request_output.finished:
208
+ requested = request_stream_intervals[int(request_id.split("-")[1])]
209
+ interval = max(requested, engine_stream_interval)
210
+ assert len(new_tokens) == interval, f"{len(new_tokens)=}, {interval=}"
211
+
212
+ for idx in range(num_requests):
213
+ request_id = f"request-{idx}"
214
+ assert gen_strings[request_id] == dummy_test_vectors.generation_strings[idx]
215
+ assert gen_tokens[request_id] == dummy_test_vectors.generation_tokens[idx]
216
+
217
+ assert not output_processor.has_unfinished_requests()
218
+
219
+
220
def _validate_logprobs(
221
gen_tokens: dict[str, list[int]],
222
gen_logprobs: dict[str, SampleLogprobs | None],
vllm/entrypoints/openai/chat_completion/protocol.py
+11
@@ -476,6 +476,16 @@ class ChatCompletionRequest(OpenAIBaseModel):
476
"can detect such behavior and terminate early, saving time and tokens.",
477
)
478
479
+ stream_interval: Annotated[int, Field(ge=1)] | None = Field(
480
+ default=None,
481
+ description=(
482
+ "Number of tokens to batch into each streamed chunk. Raises the "
483
+ "server's `--stream-interval` for this request. Values below the "
484
+ "server setting are clamped up to it. The first and last chunks "
485
+ "are always sent immediately. Ignored for non-streaming requests."
486
+ ),
487
+ )
488
+
489
# --8<-- [end:chat-completion-extra-params]
490
491
@model_validator(mode="before")
@@ -690,6 +700,7 @@ class ChatCompletionRequest(OpenAIBaseModel):
700
output_kind=(
701
RequestOutputKind.DELTA if self.stream else RequestOutputKind.FINAL_ONLY
702
),
703
+ stream_interval=self.stream_interval,
704
structured_outputs=self.extract_structured_outputs(),
705
logit_bias=self.logit_bias,
706
bad_words=self.bad_words,
vllm/entrypoints/openai/completion/protocol.py
+11
@@ -233,6 +233,16 @@ class CompletionRequest(OpenAIBaseModel):
233
),
234
)
235
236
+ stream_interval: Annotated[int, Field(ge=1)] | None = Field(
237
+ default=None,
238
+ description=(
239
+ "Number of tokens to batch into each streamed chunk. Raises the "
240
+ "server's `--stream-interval` for this request. Values below the "
241
+ "server setting are clamped up to it. The first and last chunks "
242
+ "are always sent immediately. Ignored for non-streaming requests."
243
+ ),
244
+ )
245
+
246
# --8<-- [end:completion-extra-params]
247
248
def build_tok_params(self, model_config: ModelConfig) -> TokenizeParams:
@@ -365,6 +375,7 @@ class CompletionRequest(OpenAIBaseModel):
375
output_kind=RequestOutputKind.DELTA
376
if self.stream
377
else RequestOutputKind.FINAL_ONLY,
378
+ stream_interval=self.stream_interval,
379
structured_outputs=self.extract_structured_outputs(),
380
logit_bias=self.logit_bias,
381
allowed_token_ids=self.allowed_token_ids,
vllm/sampling_params.py
+13
@@ -299,6 +299,11 @@ class SamplingParams(
299
include_stop_str_in_output: bool = False
300
"""Whether to include the stop strings in output text."""
301
output_kind: RequestOutputKind = RequestOutputKind.CUMULATIVE
302
+ stream_interval: int | None = None
303
+ """Number of newly generated tokens to batch into each streamed
304
+ `RequestOutput`. Raises the interval above the engine-level
305
+ `--stream-interval`. Values below engine setting are clamped up to it.
306
+ The first and final outputs are always emitted immediately."""
307
skip_clone: bool = False
308
"""Internal flag indicating that this SamplingParams instance is safe to
309
reuse without cloning. When True, clone() will return self without
@@ -377,6 +382,7 @@ class SamplingParams(
382
skip_special_tokens: bool = True,
383
spaces_between_special_tokens: bool = True,
384
output_kind: RequestOutputKind = RequestOutputKind.CUMULATIVE,
385
+ stream_interval: int | None = None,
386
structured_outputs: StructuredOutputsParams | None = None,
387
logit_bias: dict[int, float] | dict[str, float] | None = None,
388
allowed_token_ids: list[int] | None = None,
@@ -439,6 +445,7 @@ class SamplingParams(
445
skip_special_tokens=skip_special_tokens,
446
spaces_between_special_tokens=spaces_between_special_tokens,
447
output_kind=output_kind,
448
+ stream_interval=stream_interval,
449
structured_outputs=structured_outputs,
450
logit_bias=logit_bias,
451
allowed_token_ids=allowed_token_ids,
@@ -585,6 +592,12 @@ class SamplingParams(
592
f"min_tokens must be less than or equal to "
593
f"max_tokens={self.max_tokens}, got {self.min_tokens}."
594
)
595
+ if self.stream_interval is not None and self.stream_interval < 1:
596
+ raise VLLMValidationError(
597
+ f"stream_interval must be at least 1, got {self.stream_interval}.",
598
+ parameter="stream_interval",
599
+ value=self.stream_interval,
600
+ )
601
if self.logprobs is not None and self.logprobs != -1 and self.logprobs < 0:
602
raise VLLMValidationError(
603
f"logprobs must be non-negative or -1, got {self.logprobs}.",
vllm/v1/engine/output_processor.py
+3
@@ -224,6 +224,9 @@ class RequestState:
224
if not sampling_params.detokenize:
225
tokenizer = None
226
output_kind = sampling_params.output_kind
227
+ if sampling_params.stream_interval is not None:
228
+ # clamp to the engine-level stream interval.
229
+ stream_interval = max(sampling_params.stream_interval, stream_interval)
230
logprobs_processor = LogprobsProcessor.from_new_request(
231
tokenizer=tokenizer,
232
request=request,