main
py 607 lines 25.5 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 import time
5 from collections.abc import Callable, Mapping
6 from functools import partial
7 from typing import Any, Literal
8
9 import vllm.envs as envs
10 from vllm.config import VllmConfig
11 from vllm.exceptions import VLLMValidationError
12 from vllm.inputs import (
13 EngineInput,
14 PromptType,
15 SingletonInput,
16 split_enc_dec_input,
17 )
18 from vllm.logger import init_logger
19 from vllm.lora.request import LoRARequest
20 from vllm.multimodal import MULTIMODAL_REGISTRY, MultiModalRegistry
21 from vllm.multimodal.encoder_budget import MultiModalBudget
22 from vllm.multimodal.inputs import MultiModalFeatureSpec
23 from vllm.multimodal.utils import argsort_mm_positions
24 from vllm.platforms import current_platform
25 from vllm.pooling_params import PoolingParams
26 from vllm.renderers import BaseRenderer, renderer_from_config
27 from vllm.renderers.inputs.preprocess import parse_model_prompt
28 from vllm.sampling_params import SamplingParams
29 from vllm.tasks import GENERATION_TASKS, POOLING_TASKS, SupportedTask
30 from vllm.tokenizers import TokenizerLike
31 from vllm.utils import length_from_prompt_token_ids_or_embeds, random_uuid
32 from vllm.utils.async_utils import make_async
33 from vllm.utils.diffusion import validate_diffusion_sampling_params
34 from vllm.utils.jsontree import json_iter_leaves
35 from vllm.v1.engine import EngineCoreRequest
36 from vllm.v1.kv_hints import KvHintsEnvelope
37
38 logger = init_logger(__name__)
39
40
41 class InputProcessor:
42 def __init__(
43 self,
44 vllm_config: VllmConfig,
45 renderer: BaseRenderer | None = None,
46 *,
47 mm_registry: MultiModalRegistry = MULTIMODAL_REGISTRY,
48 ) -> None:
49 self.vllm_config = vllm_config
50 self.model_config = model_config = vllm_config.model_config
51 self.cache_config = vllm_config.cache_config
52 self.lora_config = vllm_config.lora_config
53 self.scheduler_config = vllm_config.scheduler_config
54 self.speculative_config = vllm_config.speculative_config
55 self.structured_outputs_config = vllm_config.structured_outputs_config
56 self.observability_config = vllm_config.observability_config
57 self.diffusion_config = vllm_config.diffusion_config
58 # Load the custom logits processor classes once; the returned callable
59 # runs their validate_params hooks per request at admission.
60 self.validate_logits_processors_params = (
61 self._build_logits_processors_params_validator()
62 )
63
64 self.generation_config_fields = model_config.try_get_generation_config()
65
66 self.renderer = renderer or renderer_from_config(vllm_config)
67
68 self.supports_mm_inputs = model_config.supports_multimodal_inputs
69 self.mm_encoder_cache_size = 0
70 self.skip_prompt_length_check = False
71 if self.supports_mm_inputs:
72 mm_budget = MultiModalBudget(vllm_config, mm_registry)
73 self.mm_encoder_cache_size = mm_budget.encoder_cache_size
74 self.skip_prompt_length_check = (
75 mm_budget.processor.info.skip_prompt_length_check
76 )
77 mm_budget.reset_cache() # Not used anymore
78
79 # Raw-prompt preprocessing (tokenization and multimodal processing)
80 # is blocking, so async callers should run it on the renderer's
81 # thread pool to keep their event loop responsive.
82 self.process_inputs_async = make_async(
83 self.process_inputs, executor=self.renderer._executor
84 )
85
86 @property
87 def tokenizer(self) -> TokenizerLike | None:
88 return self.renderer.tokenizer
89
90 def get_tokenizer(self) -> TokenizerLike:
91 return self.renderer.get_tokenizer()
92
93 def _build_logits_processors_params_validator(
94 self,
95 ) -> Callable[[SamplingParams], None]:
96 """Load the custom logits processor classes and return the per-request
97 params validator for the active model runner."""
98 custom_logitsprocs = self.model_config.logits_processors
99 if self.vllm_config.use_v2_model_runner:
100 if self.model_config.runner_type == "pooling" and not custom_logitsprocs:
101 return lambda _: None
102
103 from vllm.v1.worker.gpu.sample.logits_processor import (
104 build_custom_logits_processors_params_validator,
105 )
106
107 return build_custom_logits_processors_params_validator(custom_logitsprocs)
108
109 from vllm.v1.sample.logits_processor import (
110 validate_logits_processors_parameters,
111 )
112
113 return partial(validate_logits_processors_parameters, custom_logitsprocs)
114
115 def _validate_params(
116 self,
117 params: SamplingParams | PoolingParams,
118 supported_tasks: tuple[SupportedTask, ...],
119 ) -> None:
120 """Raise `ValueError` if SamplingParams or PoolingParams is not valid."""
121 if isinstance(params, SamplingParams):
122 supported_generation_tasks = [
123 task for task in supported_tasks if task in GENERATION_TASKS
124 ]
125 if not supported_generation_tasks:
126 raise VLLMValidationError("This model does not support generation")
127
128 params.verify(
129 self.model_config,
130 self.speculative_config,
131 self.structured_outputs_config,
132 self.tokenizer,
133 )
134 if params.prompt_logprob_token_ids is not None:
135 if not self.vllm_config.use_v2_model_runner:
136 raise VLLMValidationError(
137 "prompt_logprob_token_ids requires the V2 model runner "
138 "(VLLM_USE_V2_MODEL_RUNNER=1).",
139 parameter="prompt_logprob_token_ids",
140 )
141 if self.vllm_config.cache_config.kv_sharing_fast_prefill:
142 raise VLLMValidationError(
143 "prompt_logprob_token_ids is incorrect with "
144 "--kv-sharing-fast-prefill; disable it for scoring.",
145 parameter="prompt_logprob_token_ids",
146 )
147
148 self.validate_logits_processors_params(params)
149
150 if self.model_config.is_diffusion:
151 # Without --diffusion-config the served canvas is unknown here;
152 # the ids and the read-only normalisation are still checked.
153 validate_diffusion_sampling_params(
154 params,
155 canvas_length=(
156 self.diffusion_config.canvas_length
157 if self.diffusion_config is not None
158 else None
159 ),
160 vocab_size=self.model_config.get_vocab_size(),
161 )
162
163 if self.model_config.return_sampling_mask:
164 if params.temperature <= 0:
165 raise ValueError(
166 "sampling distribution replay requires temperature > 0"
167 )
168 if params.top_k <= 0:
169 raise ValueError(
170 "sampling distribution replay requires top_k > 0 to "
171 "bound sampling mask size, reduce transfer overhead, "
172 "and avoid potential OOMs"
173 )
174 if params.thinking_token_budget is not None and (
175 self.vllm_config.reasoning_config is None
176 or not self.vllm_config.reasoning_config.enabled
177 ):
178 raise VLLMValidationError(
179 "thinking_token_budget is set but reasoning_config is "
180 "not configured. Please set --reasoning-parser "
181 "and/or --reasoning-config to use thinking_token_budget."
182 )
183 if (
184 params.trace_decode_token_ids
185 and not self.model_config.enable_trace_replay
186 ):
187 raise VLLMValidationError(
188 "trace_decode_token_ids is set but trace replay is not "
189 "enabled. Start the engine with --enable-trace-replay "
190 "to use it."
191 )
192 elif isinstance(params, PoolingParams):
193 supported_pooling_tasks = [
194 task for task in supported_tasks if task in POOLING_TASKS
195 ]
196 if not supported_pooling_tasks:
197 raise VLLMValidationError("This model does not support pooling")
198
199 if params.task is None:
200 if "token_embed" in supported_pooling_tasks:
201 params.task = "token_embed"
202 elif "token_classify" in supported_pooling_tasks:
203 params.task = "token_classify"
204 elif "plugin" in supported_pooling_tasks:
205 params.task = "plugin"
206
207 if params.task not in supported_pooling_tasks:
208 raise VLLMValidationError(
209 f"Unsupported task: {params.task!r} "
210 f"Supported tasks: {supported_pooling_tasks}"
211 )
212
213 params.verify(self.model_config)
214 else:
215 raise TypeError(
216 f"params must be either SamplingParams or PoolingParams, "
217 f"but got {type(params).__name__}"
218 )
219
220 def _normalize_trace_replay_params(
221 self, sampling_params: SamplingParams, prompt_len: int
222 ) -> None:
223 """Apply trace replay's generation semantics to request-local params."""
224 trace_token_ids = sampling_params.trace_decode_token_ids
225 assert trace_token_ids
226 assert sampling_params.max_tokens is not None
227
228 max_trace_len = max(self.model_config.max_model_len - prompt_len, 1)
229 trace_token_ids = trace_token_ids[:max_trace_len]
230 sampling_params.trace_decode_token_ids = trace_token_ids
231
232 # Apply this after the generation config so its EOS token cannot stop
233 # replay before the trace is exhausted.
234 sampling_params.max_tokens = min(
235 len(trace_token_ids), sampling_params.max_tokens
236 )
237 sampling_params.min_tokens = 0
238 sampling_params.ignore_eos = True
239 sampling_params._eos_token_id = None
240 sampling_params.stop = []
241 sampling_params.stop_token_ids = []
242 sampling_params._all_stop_token_ids = set()
243
244 def _validate_lora(self, lora_request: LoRARequest | None) -> None:
245 if lora_request is None:
246 return
247
248 # LoRA request passed in while LoRA is not enabled
249 if not self.lora_config:
250 raise VLLMValidationError(
251 f"Got lora_request {lora_request} but LoRA is not enabled!"
252 )
253
254 if self.tokenizer is not None:
255 logger.warning_once(
256 "vLLM has deprecated support for supporting different "
257 "tokenizers for different LoRAs. By default, vLLM uses base "
258 "model's tokenizer. If you are using a LoRA "
259 "with its own tokenizer, consider specifying `--tokenizer "
260 "[lora_path]` to use the LoRA tokenizer."
261 )
262
263 def _get_mm_identifier(
264 self,
265 mm_hash: str,
266 lora_request: LoRARequest | None,
267 ) -> str:
268 """When enable_tower_connector_lora is True, multi-modal embeddings
269 vary depending on the LoRA request. Therefore, the mm_hash must be
270 generated based on the LoRA request to prevent incorrect cache hits.
271 """
272 if (
273 lora_request is None
274 or self.lora_config is None
275 or not self.lora_config.enable_tower_connector_lora
276 ):
277 return mm_hash
278 return f"{lora_request.lora_name}:{mm_hash}"
279
280 def inject_into_mm_cache(
281 self,
282 mm_hashes: dict[str, list[str]],
283 mm_kwargs: dict[str, list],
284 ) -> None:
285 """Inject pre-processed mm_kwargs into the processor cache.
286
287 Call this when mm_kwargs have already been through the HF processor
288 externally (e.g. by a frontend that transfers pre-processed tensors
289 to the backend). This ensures MM cache hit rate metrics are reported
290 accurately and avoids redundant processing on subsequent requests
291 with the same images.
292
293 Uses ``get_and_update_item()`` with an empty prompt_updates list,
294 since token expansion has already been handled externally.
295 """
296 cache = self.renderer.mm_processor_cache
297 if cache is None:
298 return
299 try:
300 for modality, hashes in mm_hashes.items():
301 items = mm_kwargs.get(modality, [])
302 for i, mm_hash in enumerate(hashes):
303 if i < len(items) and items[i] is not None:
304 # Insert into cache via get_and_update_item.
305 # Use the returned item (may be an address for SHM
306 # cache or the original item for LRU cache).
307 items[i], _ = cache.get_and_update_item(
308 (items[i], []),
309 mm_hash,
310 )
311 # Update cache stats to reflect the externally processed items
312 self.renderer.update_mm_cache_stats()
313 except Exception:
314 logger.warning(
315 "Failed to inject mm_kwargs into processor cache",
316 exc_info=True,
317 )
318
319 @staticmethod
320 def assign_request_id(request: EngineCoreRequest):
321 """Replace the externally supplied request ID with an internal request ID
322 that adds 8 random characters in order to ensure uniqueness.
323 """
324 if request.external_req_id is not None:
325 raise ValueError(
326 "The external_req_id field should not be set on EngineCoreRequests"
327 " passed to vLLM; use the request_id field."
328 )
329 request.external_req_id = request.request_id
330 if envs.VLLM_DISABLE_REQUEST_ID_RANDOMIZATION:
331 logger.warning_once(
332 "VLLM_DISABLE_REQUEST_ID_RANDOMIZATION is set and will be "
333 "removed in a future release. Duplicate externally-provided "
334 "request IDs may cause failures and/or subtle correctness errors."
335 )
336 else:
337 request.request_id = f"{request.external_req_id}-{random_uuid():.8}"
338
339 def process_inputs(
340 self,
341 request_id: str,
342 prompt: PromptType | EngineInput,
343 params: SamplingParams | PoolingParams,
344 supported_tasks: tuple[SupportedTask, ...],
345 arrival_time: float | None = None,
346 lora_request: LoRARequest | None = None,
347 tokenization_kwargs: dict[str, Any] | None = None,
348 trace_headers: Mapping[str, str] | None = None,
349 priority: int = 0,
350 data_parallel_rank: int | None = None,
351 resumable: bool = False,
352 session_id: str | None = None,
353 kv_hints: KvHintsEnvelope | None = None,
354 ) -> EngineCoreRequest:
355 self._validate_params(params, supported_tasks)
356 self._validate_lora(lora_request)
357
358 parallel_config = self.vllm_config.parallel_config
359 dp_size = parallel_config.data_parallel_size
360 dp_local_size = parallel_config.data_parallel_size_local
361 num_ranks = dp_local_size if parallel_config.local_engines_only else dp_size
362 if data_parallel_rank is not None and not (0 <= data_parallel_rank < num_ranks):
363 raise VLLMValidationError(
364 f"data_parallel_rank {data_parallel_rank} "
365 f"is out of range [0, {num_ranks})."
366 )
367
368 if isinstance(prompt, dict) and "type" in prompt:
369 if arrival_time is None:
370 arrival_time = prompt.get("arrival_time", time.time()) # type: ignore[assignment]
371
372 engine_input: EngineInput = prompt # type: ignore[assignment]
373 else:
374 logger.warning_once(
375 "Passing raw prompts to InputProcessor is deprecated "
376 "and will be removed in the future. You should instead pass "
377 "the outputs of Renderer.render_cmpl() or Renderer.render_chat()."
378 )
379
380 if arrival_time is None:
381 arrival_time = time.time()
382
383 renderer = self.renderer
384 model_config = self.model_config
385
386 parsed_prompt = parse_model_prompt(model_config, prompt)
387 tok_params = renderer.default_cmpl_tok_params.with_kwargs(
388 **(tokenization_kwargs or {})
389 )
390
391 (engine_input,) = renderer.render_cmpl(
392 [parsed_prompt],
393 tok_params,
394 )
395
396 current_platform.validate_request(engine_input, params)
397
398 encoder_input, decoder_input = split_enc_dec_input(engine_input)
399 self._validate_model_inputs(encoder_input, decoder_input)
400
401 # Mypy can be conservative for TypedDict unions; normalize access.
402 if decoder_input["type"] == "embeds":
403 prompt_embeds = decoder_input["prompt_embeds"]
404 prompt_token_ids = decoder_input.get("prompt_token_ids")
405 prompt_is_token_ids = decoder_input.get("is_token_ids")
406 else:
407 prompt_token_ids = decoder_input["prompt_token_ids"]
408 prompt_embeds = None
409 prompt_is_token_ids = None
410
411 sampling_params = None
412 pooling_params = None
413 if isinstance(params, SamplingParams):
414 # TODO: can we avoid cloning here in multiproc case?
415 sampling_params = params.clone()
416 prompt_len = length_from_prompt_token_ids_or_embeds(
417 prompt_token_ids, prompt_embeds
418 )
419 if not 0 <= sampling_params.routed_experts_prompt_start <= prompt_len:
420 raise VLLMValidationError(
421 f"routed_experts_prompt_start must be between 0 and "
422 f"the prompt length ({prompt_len}), inclusive.",
423 parameter="routed_experts_prompt_start",
424 value=sampling_params.routed_experts_prompt_start,
425 )
426 # If unset max tokens, then generate up to the max_model_len.
427 if sampling_params.max_tokens is None:
428 sampling_params.max_tokens = (
429 self.model_config.max_model_len - prompt_len
430 )
431 # min_tokens is not checked while max_tokens is unset.
432 if sampling_params.min_tokens > sampling_params.max_tokens:
433 raise VLLMValidationError(
434 f"min_tokens must be less than or equal to "
435 f"max_tokens={sampling_params.max_tokens}, got "
436 f"{sampling_params.min_tokens}."
437 )
438
439 sampling_params.update_from_generation_config(
440 self.generation_config_fields,
441 self.renderer.get_eos_token_id(),
442 )
443 if self.tokenizer is not None:
444 sampling_params.update_from_tokenizer(self.tokenizer)
445 if sampling_params.trace_decode_token_ids:
446 self._normalize_trace_replay_params(sampling_params, prompt_len)
447 else:
448 pooling_params = params.clone()
449
450 # Multimodal related.
451 mm_features: list[MultiModalFeatureSpec] | None = None
452
453 if decoder_input["type"] == "multimodal":
454 decoder_mm_inputs = decoder_input["mm_kwargs"]
455 decoder_mm_positions = decoder_input["mm_placeholders"]
456 decoder_mm_hashes = decoder_input["mm_hashes"]
457
458 if not all(
459 isinstance(leaf, str) for leaf in json_iter_leaves(decoder_mm_hashes)
460 ):
461 raise ValueError(
462 f"mm_hashes must contain only strings, got: {decoder_mm_hashes}. "
463 "This is likely due to an incorrect custom implementation of "
464 "MultiModalProcessor.apply method."
465 )
466
467 # Merge and flatten multimodal placeholders, hashes and inputs
468 # from dictionaries to lists, and sort them by each item's position
469 # in the input sequence.
470 sorted_mm_idxs = argsort_mm_positions(decoder_mm_positions)
471
472 mm_features = []
473 for modality, idx in sorted_mm_idxs:
474 base_mm_hash = decoder_mm_hashes[modality][idx]
475 mm_features.append(
476 MultiModalFeatureSpec(
477 data=decoder_mm_inputs[modality][idx],
478 modality=modality,
479 identifier=self._get_mm_identifier(
480 base_mm_hash,
481 lora_request,
482 ),
483 mm_position=decoder_mm_positions[modality][idx],
484 mm_hash=base_mm_hash,
485 )
486 )
487
488 return EngineCoreRequest(
489 request_id=request_id,
490 prompt_token_ids=prompt_token_ids,
491 prompt_embeds=prompt_embeds,
492 prompt_is_token_ids=prompt_is_token_ids,
493 mm_features=mm_features,
494 sampling_params=sampling_params,
495 pooling_params=pooling_params,
496 arrival_time=arrival_time,
497 lora_request=lora_request,
498 cache_salt=decoder_input.get("cache_salt"),
499 priority=priority,
500 data_parallel_rank=data_parallel_rank,
501 trace_headers=trace_headers,
502 resumable=resumable,
503 session_id=session_id,
504 kv_hints=kv_hints,
505 )
506
507 def _validate_prompt_len(
508 self,
509 prompt_len: int,
510 prompt_type: Literal["encoder", "decoder"],
511 ):
512 if self.skip_prompt_length_check and prompt_type == "encoder":
513 return
514
515 if prompt_len == 0 and prompt_type == "decoder":
516 raise VLLMValidationError(f"The {prompt_type} prompt cannot be empty")
517
518 model_config = self.model_config
519 max_prompt_len = (
520 model_config.max_model_len
521 if prompt_type == "decoder"
522 else self.mm_encoder_cache_size
523 )
524 if prompt_len > max_prompt_len:
525 if self.supports_mm_inputs:
526 suggestion = (
527 "Make sure that `max_model_len` is no smaller than the "
528 "number of text tokens plus multimodal tokens. For image "
529 "inputs, the number of image tokens depends on the number "
530 "of images, and possibly their aspect ratios as well."
531 )
532 else:
533 suggestion = (
534 "Make sure that `max_model_len` is no smaller than the "
535 "number of text tokens."
536 )
537
538 raise VLLMValidationError(
539 f"The {prompt_type} prompt (length {prompt_len}) is "
540 f"longer than the maximum model length of {max_prompt_len}. "
541 f"{suggestion}"
542 )
543 elif prompt_len == max_prompt_len and model_config.runner_type == "generate":
544 suggestion = (
545 "Make sure that `max_model_len` is no smaller than the "
546 "number of text tokens (prompt + requested output tokens)."
547 )
548 raise VLLMValidationError(
549 f"The {prompt_type} prompt (length {prompt_len}) plus the number of "
550 f"requested output tokens (at least 1) is longer than the maximum "
551 f"model length of {max_prompt_len}. {suggestion}"
552 )
553
554 def _validate_model_input(
555 self,
556 prompt_input: SingletonInput,
557 prompt_type: Literal["encoder", "decoder"],
558 ) -> None:
559 prompt_ids = (
560 None
561 if prompt_input["type"] == "embeds"
562 else prompt_input["prompt_token_ids"]
563 )
564 prompt_embeds = (
565 prompt_input["prompt_embeds"] if prompt_input["type"] == "embeds" else None
566 )
567
568 prompt_len = length_from_prompt_token_ids_or_embeds(prompt_ids, prompt_embeds)
569 self._validate_prompt_len(prompt_len, prompt_type)
570
571 if prompt_input["type"] == "embeds":
572 is_token_ids = prompt_input.get("is_token_ids")
573 if is_token_ids is not None and len(is_token_ids) != prompt_len:
574 raise VLLMValidationError(
575 "prompt_is_token_ids must have the same length as prompt_embeds "
576 f"(expected {prompt_len}, got {len(is_token_ids)}).",
577 parameter="prompt_is_token_ids",
578 )
579
580 if prompt_input["type"] == "multimodal":
581 decoder_mm_positions = prompt_input["mm_placeholders"]
582 for modality, mm_positions in decoder_mm_positions.items():
583 for mm_position in mm_positions:
584 num_embeds = mm_position.get_num_embeds()
585 if num_embeds > self.mm_encoder_cache_size:
586 raise VLLMValidationError(
587 f"The {prompt_type} prompt contains a(n) {modality} item "
588 f"with {num_embeds} embedding tokens, which exceeds the "
589 f"pre-allocated encoder cache size "
590 f"{self.mm_encoder_cache_size}. Please reduce the input "
591 f"size or increase the encoder cache size "
592 f"by setting --limit-mm-per-prompt at startup."
593 )
594
595 # Shared by generate, embedding and pooling requests.
596 if prompt_ids:
597 self.renderer.validate_token_ids(prompt_ids)
598
599 def _validate_model_inputs(
600 self,
601 encoder_input: SingletonInput | None,
602 decoder_input: SingletonInput,
603 ):
604 if encoder_input is not None:
605 self._validate_model_input(encoder_input, prompt_type="encoder")
606
607 self._validate_model_input(decoder_input, prompt_type="decoder")