main
py 634 lines 23.6 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 from collections.abc import Callable, Iterable, Sequence
5 from typing import Any
6
7 from tqdm import tqdm
8 from typing_extensions import TypeVar
9
10 from vllm import (
11 PoolingParams,
12 PoolingRequestOutput,
13 PromptType,
14 RequestOutput,
15 SamplingParams,
16 )
17 from vllm.config import ModelConfig
18 from vllm.entrypoints.chat_utils import (
19 ChatCompletionMessageParam,
20 ChatTemplateContentFormatOption,
21 )
22 from vllm.exceptions import VLLMValidationError
23 from vllm.inputs import EngineInput
24 from vllm.logger import init_logger
25 from vllm.lora.request import LoRARequest
26 from vllm.renderers import BaseRenderer, ChatParams, merge_kwargs
27 from vllm.renderers.inputs.preprocess import (
28 conversation_to_seq,
29 parse_model_prompt,
30 prompt_to_seq,
31 )
32 from vllm.sampling_params import RequestOutputKind
33 from vllm.utils.counter import Counter
34 from vllm.utils.mistral import is_mistral_tokenizer
35 from vllm.utils.tqdm_utils import maybe_tqdm
36 from vllm.v1.engine.llm_engine import LLMEngine
37
38 logger = init_logger(__name__)
39
40
41 _P = TypeVar("_P", bound=SamplingParams | PoolingParams | None)
42 _O = TypeVar(
43 "_O",
44 bound=RequestOutput | PoolingRequestOutput,
45 default=RequestOutput | PoolingRequestOutput,
46 )
47 _R = TypeVar("_R", default=Any)
48
49
50 class OfflineInferenceMixin:
51 """Offline inference utils."""
52
53 request_counter: Counter
54 renderer: BaseRenderer
55 llm_engine: "LLMEngine"
56 model_config: ModelConfig
57
58 def _resolve_mm_lora(
59 self,
60 prompt: EngineInput,
61 lora_request: LoRARequest | None,
62 ) -> LoRARequest | None:
63 if prompt["type"] != "multimodal":
64 return lora_request
65
66 lora_config = self.llm_engine.vllm_config.lora_config
67 default_mm_loras = None if lora_config is None else lora_config.default_mm_loras
68 if not default_mm_loras:
69 return lora_request
70
71 prompt_modalities = prompt["mm_placeholders"].keys()
72 intersection = set(prompt_modalities).intersection(default_mm_loras.keys())
73 if not intersection:
74 return lora_request
75
76 if len(intersection) > 1:
77 # TODO: Would be nice to be able to have multiple loras per prompt
78 logger.warning(
79 "Multiple modality specific loras were registered and would be "
80 "used by a single prompt consuming several modalities; "
81 "currently we only support one lora per request; as such, "
82 "lora(s) registered with modalities: %s will be skipped",
83 intersection,
84 )
85 return lora_request
86
87 # Build the LoRA request; the ID of the default mm lora is the
88 # index of the modality name sorted alphabetically + 1.
89 modality_name = intersection.pop()
90 modality_lora_path = default_mm_loras[modality_name]
91 modality_lora_id = sorted(default_mm_loras).index(modality_name) + 1
92
93 # If we have a collision, warn if there is a collision,
94 # but always send the explicitly provided request.
95 if lora_request:
96 if lora_request.lora_int_id != modality_lora_id:
97 logger.warning(
98 "A modality with a registered lora and a lora_request "
99 "with a different ID were provided; falling back to the "
100 "lora_request as we only apply one LoRARequest per prompt"
101 )
102 return lora_request
103
104 return LoRARequest(
105 modality_name,
106 modality_lora_id,
107 modality_lora_path,
108 )
109
110 def _preprocess_cmpl(
111 self,
112 prompts: Sequence[PromptType],
113 tokenization_kwargs: dict[str, Any] | None = None,
114 mm_processor_kwargs: dict[str, Any] | None = None,
115 ) -> Sequence[EngineInput]:
116 """Convert prompt inputs from LLM APIs (other than [LLM.chat][]) into
117 a format that can be passed to `_add_request`.
118
119 Refer to [LLM.generate][] for a complete description of the arguments.
120
121 Returns:
122 A list of `EngineInput` objects ready to be passed into LLMEngine.
123
124 """
125 renderer = self.renderer
126 model_config = self.model_config
127
128 parsed_prompts = [
129 parse_model_prompt(model_config, prompt) for prompt in prompts
130 ]
131 tok_params = renderer.default_cmpl_tok_params.with_kwargs(
132 **(tokenization_kwargs or {})
133 )
134 prompt_extras = (
135 None
136 if mm_processor_kwargs is None
137 else {"mm_processor_kwargs": mm_processor_kwargs}
138 )
139
140 return renderer.render_cmpl(
141 parsed_prompts,
142 tok_params,
143 prompt_extras=prompt_extras,
144 )
145
146 def _preprocess_cmpl_one(
147 self,
148 prompt: PromptType,
149 tokenization_kwargs: dict[str, Any] | None = None,
150 mm_processor_kwargs: dict[str, Any] | None = None,
151 ) -> EngineInput:
152 (engine_input,) = self._preprocess_cmpl(
153 [prompt],
154 tokenization_kwargs,
155 mm_processor_kwargs=mm_processor_kwargs,
156 )
157 return engine_input
158
159 def _preprocess_chat(
160 self,
161 conversations: Sequence[list[ChatCompletionMessageParam]],
162 chat_template: str | None = None,
163 chat_template_content_format: ChatTemplateContentFormatOption = "auto",
164 chat_template_kwargs: dict[str, Any] | None = None,
165 add_generation_prompt: bool = True,
166 continue_final_message: bool = False,
167 tools: list[dict[str, Any]] | None = None,
168 tokenization_kwargs: dict[str, Any] | None = None,
169 mm_processor_kwargs: dict[str, Any] | None = None,
170 ) -> Sequence[EngineInput]:
171 """Convert a list of conversations into prompts so that they can then
172 be used as input for other LLM APIs.
173
174 Refer to [LLM.chat][] for a complete description of the arguments.
175
176 Returns:
177 A list of `EngineInput` objects ready to be passed into LLMEngine.
178
179 """
180 renderer = self.renderer
181
182 chat_params = ChatParams(
183 chat_template=chat_template,
184 chat_template_content_format=chat_template_content_format,
185 chat_template_kwargs=merge_kwargs(
186 chat_template_kwargs,
187 dict(
188 add_generation_prompt=add_generation_prompt,
189 continue_final_message=continue_final_message,
190 tools=tools,
191 tokenize=(
192 is_mistral_tokenizer(renderer.tokenizer)
193 or self.model_config.enable_prompt_embeds
194 ),
195 ),
196 ),
197 mm_processor_kwargs=mm_processor_kwargs,
198 )
199 # The chat template is responsible for emitting BOS/EOS, so do not let
200 # the tokenizer add them again unless the caller asks for it. This
201 # matches the online chat API (`ChatCompletionRequest.add_special_tokens`
202 # defaults to `False`) and avoids a double BOS for multimodal models,
203 # whose processor default is `add_special_tokens=True` (#55197).
204 tokenization_kwargs = {
205 "add_special_tokens": False,
206 **(tokenization_kwargs or {}),
207 }
208 tok_params = renderer.default_chat_tok_params.with_kwargs(**tokenization_kwargs)
209 prompt_extras = (
210 None
211 if mm_processor_kwargs is None
212 else {"mm_processor_kwargs": mm_processor_kwargs}
213 )
214
215 _, engine_inputs = renderer.render_chat(
216 conversations,
217 chat_params,
218 tok_params,
219 prompt_extras=prompt_extras,
220 )
221
222 return engine_inputs
223
224 def _preprocess_chat_one(
225 self,
226 conversation: list[ChatCompletionMessageParam],
227 chat_template: str | None = None,
228 chat_template_content_format: ChatTemplateContentFormatOption = "auto",
229 chat_template_kwargs: dict[str, Any] | None = None,
230 add_generation_prompt: bool = True,
231 continue_final_message: bool = False,
232 tools: list[dict[str, Any]] | None = None,
233 tokenization_kwargs: dict[str, Any] | None = None,
234 mm_processor_kwargs: dict[str, Any] | None = None,
235 ) -> EngineInput:
236 (engine_input,) = self._preprocess_chat(
237 [conversation],
238 chat_template=chat_template,
239 chat_template_content_format=chat_template_content_format,
240 chat_template_kwargs=chat_template_kwargs,
241 add_generation_prompt=add_generation_prompt,
242 continue_final_message=continue_final_message,
243 tools=tools,
244 tokenization_kwargs=tokenization_kwargs,
245 mm_processor_kwargs=mm_processor_kwargs,
246 )
247
248 return engine_input
249
250 def _params_to_seq(
251 self,
252 params: _P | Sequence[_P],
253 num_requests: int,
254 ) -> Sequence[_P]:
255 if isinstance(params, Sequence):
256 if len(params) != num_requests:
257 raise VLLMValidationError(
258 f"The lengths of prompts ({num_requests}) "
259 f"and params ({len(params)}) must be the same."
260 )
261
262 return params
263
264 return [params] * num_requests
265
266 def _lora_request_to_seq(
267 self,
268 lora_request: LoRARequest | None | Sequence[LoRARequest | None],
269 num_requests: int,
270 ) -> Sequence[LoRARequest | None]:
271 if isinstance(lora_request, Sequence):
272 if len(lora_request) != num_requests:
273 raise VLLMValidationError(
274 f"The lengths of prompts ({num_requests}) "
275 f"and lora_request ({len(lora_request)}) must be the same."
276 )
277
278 return lora_request
279
280 return [lora_request] * num_requests
281
282 def _priority_to_seq(
283 self,
284 priority: list[int] | None,
285 num_requests: int,
286 ) -> Sequence[int]:
287 if priority is not None:
288 if len(priority) != num_requests:
289 raise VLLMValidationError(
290 f"The lengths of prompts ({num_requests}) "
291 f"and priority ({len(priority)}) must be the same."
292 )
293
294 return priority
295
296 return [0] * num_requests
297
298 def _add_completion_requests(
299 self,
300 prompts: PromptType | Sequence[PromptType],
301 params: SamplingParams
302 | PoolingParams
303 | Sequence[SamplingParams | PoolingParams],
304 *,
305 use_tqdm: bool | Callable[..., tqdm] = True,
306 lora_request: Sequence[LoRARequest] | LoRARequest | None = None,
307 priority: list[int] | None = None,
308 tokenization_kwargs: dict[str, Any] | None = None,
309 mm_processor_kwargs: dict[str, Any] | None = None,
310 ) -> list[str]:
311 seq_prompts = prompt_to_seq(prompts)
312 seq_params = self._params_to_seq(params, len(seq_prompts))
313 seq_lora_requests = self._lora_request_to_seq(lora_request, len(seq_prompts))
314 seq_priority = self._priority_to_seq(priority, len(seq_prompts))
315
316 return self._render_and_add_requests(
317 prompts=(
318 self._preprocess_cmpl_one(
319 prompt,
320 tokenization_kwargs,
321 mm_processor_kwargs=mm_processor_kwargs,
322 )
323 for prompt in maybe_tqdm(
324 seq_prompts,
325 use_tqdm=use_tqdm,
326 desc="Rendering prompts",
327 )
328 ),
329 params=seq_params,
330 lora_requests=seq_lora_requests,
331 priorities=seq_priority,
332 )
333
334 def _run_completion(
335 self,
336 prompts: PromptType | Sequence[PromptType],
337 params: SamplingParams
338 | PoolingParams
339 | Sequence[SamplingParams | PoolingParams],
340 output_type: type[_O],
341 *,
342 use_tqdm: bool | Callable[..., tqdm] = True,
343 lora_request: Sequence[LoRARequest] | LoRARequest | None = None,
344 priority: list[int] | None = None,
345 tokenization_kwargs: dict[str, Any] | None = None,
346 mm_processor_kwargs: dict[str, Any] | None = None,
347 ):
348 self._add_completion_requests(
349 prompts=prompts,
350 params=params,
351 use_tqdm=use_tqdm,
352 lora_request=lora_request,
353 priority=priority,
354 tokenization_kwargs=tokenization_kwargs,
355 mm_processor_kwargs=mm_processor_kwargs,
356 )
357 return self._run_engine(use_tqdm=use_tqdm, output_type=output_type)
358
359 def _run_chat(
360 self,
361 messages: list[ChatCompletionMessageParam]
362 | Sequence[list[ChatCompletionMessageParam]],
363 params: SamplingParams
364 | PoolingParams
365 | Sequence[SamplingParams | PoolingParams],
366 output_type: type[_O],
367 *,
368 use_tqdm: bool | Callable[..., tqdm] = True,
369 lora_request: Sequence[LoRARequest] | LoRARequest | None = None,
370 chat_template: str | None = None,
371 chat_template_content_format: ChatTemplateContentFormatOption = "auto",
372 add_generation_prompt: bool = True,
373 continue_final_message: bool = False,
374 tools: list[dict[str, Any]] | None = None,
375 chat_template_kwargs: dict[str, Any] | None = None,
376 tokenization_kwargs: dict[str, Any] | None = None,
377 mm_processor_kwargs: dict[str, Any] | None = None,
378 ):
379 self._add_chat_requests(
380 messages=messages,
381 params=params,
382 use_tqdm=use_tqdm,
383 lora_request=lora_request,
384 chat_template=chat_template,
385 chat_template_content_format=chat_template_content_format,
386 chat_template_kwargs=chat_template_kwargs,
387 add_generation_prompt=add_generation_prompt,
388 continue_final_message=continue_final_message,
389 tools=tools,
390 tokenization_kwargs=tokenization_kwargs,
391 mm_processor_kwargs=mm_processor_kwargs,
392 )
393 return self._run_engine(output_type=output_type, use_tqdm=use_tqdm)
394
395 def _add_chat_requests(
396 self,
397 messages: list[ChatCompletionMessageParam]
398 | Sequence[list[ChatCompletionMessageParam]],
399 params: SamplingParams
400 | PoolingParams
401 | Sequence[SamplingParams | PoolingParams],
402 *,
403 use_tqdm: bool | Callable[..., tqdm] = True,
404 lora_request: Sequence[LoRARequest] | LoRARequest | None = None,
405 priority: list[int] | None = None,
406 chat_template: str | None = None,
407 chat_template_content_format: ChatTemplateContentFormatOption = "auto",
408 add_generation_prompt: bool = True,
409 continue_final_message: bool = False,
410 tools: list[dict[str, Any]] | None = None,
411 chat_template_kwargs: dict[str, Any] | None = None,
412 tokenization_kwargs: dict[str, Any] | None = None,
413 mm_processor_kwargs: dict[str, Any] | None = None,
414 ) -> list[str]:
415 seq_convs = conversation_to_seq(messages)
416 seq_params = self._params_to_seq(params, len(seq_convs))
417 seq_lora_requests = self._lora_request_to_seq(lora_request, len(seq_convs))
418 seq_priority = self._priority_to_seq(priority, len(seq_convs))
419
420 # When thinking is enabled or tools are provided, and the model
421 # uses special tokens for structured output (e.g. Gemma4's
422 # <|channel>, <|tool_call>, <|"|>), automatically set
423 # skip_special_tokens=False so these tokens are preserved in
424 # output.text for downstream parsing.
425 needs_parsing = (
426 chat_template_kwargs and chat_template_kwargs.get("enable_thinking")
427 ) or tools
428 if needs_parsing:
429 self._adjust_params_for_parsing(seq_params)
430
431 return self._render_and_add_requests(
432 prompts=(
433 self._preprocess_chat_one(
434 conversation,
435 chat_template=chat_template,
436 chat_template_content_format=chat_template_content_format,
437 chat_template_kwargs=chat_template_kwargs,
438 add_generation_prompt=add_generation_prompt,
439 continue_final_message=continue_final_message,
440 tools=tools,
441 tokenization_kwargs=tokenization_kwargs,
442 mm_processor_kwargs=mm_processor_kwargs,
443 )
444 for conversation in maybe_tqdm(
445 seq_convs,
446 use_tqdm=use_tqdm,
447 desc="Rendering conversations",
448 )
449 ),
450 params=seq_params,
451 lora_requests=seq_lora_requests,
452 priorities=seq_priority,
453 )
454
455 def _adjust_params_for_parsing(
456 self, params: Sequence[SamplingParams | PoolingParams]
457 ) -> None:
458 """Set ``skip_special_tokens=False`` when the model encodes
459 structured output syntax as special tokens.
460
461 Models like Gemma4 register thinking delimiters
462 (``<|channel>``/``<channel|>``) and tool call tokens
463 (``<|tool_call>``/``<tool_call|>``/``<|"|>``) as special tokens.
464 The default ``skip_special_tokens=True`` strips them from
465 ``output.text``, breaking parsing of both reasoning blocks and
466 tool calls.
467
468 This is a no-op for models whose structured tokens are regular
469 text tokens (e.g. DeepSeek's ``<think>``/``</think>``).
470 """
471 # The offline API currently lacks a unified rendering pipeline.
472 # Until the planned Renderer refactor is complete, we hardcode
473 # this token preservation logic specifically for Gemma4 models
474 # to avoid regressions on other models.
475 hf_config = getattr(self.model_config, "hf_config", None)
476 architectures = getattr(hf_config, "architectures", [])
477
478 if any("Gemma4" in arch for arch in architectures):
479 tokenizer = self.renderer.get_tokenizer()
480 vocab = tokenizer.get_vocab()
481 special_ids = set(getattr(tokenizer, "all_special_ids", []))
482
483 # Tokens used for thinking delimiters and tool call syntax
484 # that some models (Gemma4) register as special tokens.
485 structured_tokens = (
486 "<|channel>",
487 "<channel|>", # thinking delimiters
488 "<|tool_call>",
489 "<tool_call|>", # tool call delimiters
490 '<|"|>', # string quoting in tool args
491 )
492 needs_special = any(
493 vocab.get(tok) in special_ids
494 for tok in structured_tokens
495 if tok in vocab
496 )
497 if needs_special:
498 for sp in params:
499 if isinstance(sp, SamplingParams) and sp.skip_special_tokens:
500 sp.skip_special_tokens = False
501
502 def _render_and_run_requests(
503 self,
504 prompts: Iterable[EngineInput],
505 params: Sequence[SamplingParams | PoolingParams],
506 output_type: type[_O],
507 *,
508 lora_requests: Sequence[LoRARequest | None] | None = None,
509 priorities: Sequence[int] | None = None,
510 use_tqdm: bool | Callable[..., tqdm] = True,
511 ):
512 if isinstance(prompts, (list, tuple)):
513 logger.warning_once(
514 "Rendering all prompts before adding them to the engine "
515 "is less efficient than performing both on the same prompt "
516 "before processing the next prompt. You should instead pass "
517 "a generator that renders one prompt per iteration, as that allows "
518 "engine execution to begin for the first prompt while processing "
519 "the next prompt."
520 )
521
522 self._render_and_add_requests(
523 prompts=prompts,
524 params=params,
525 lora_requests=lora_requests,
526 priorities=priorities,
527 )
528
529 return self._run_engine(output_type, use_tqdm=use_tqdm)
530
531 def _render_and_add_requests(
532 self,
533 prompts: Iterable[EngineInput],
534 params: Sequence[SamplingParams | PoolingParams],
535 *,
536 lora_requests: Sequence[LoRARequest | None] | None = None,
537 priorities: Sequence[int] | None = None,
538 ) -> list[str]:
539 added_request_ids: list[str] = []
540
541 try:
542 for i, prompt in enumerate(prompts):
543 request_id = self._add_request(
544 prompt,
545 params[i],
546 lora_request=self._resolve_mm_lora(
547 prompt,
548 None if lora_requests is None else lora_requests[i],
549 ),
550 priority=0 if priorities is None else priorities[i],
551 )
552 added_request_ids.append(request_id)
553 except Exception as e:
554 if added_request_ids:
555 self.llm_engine.abort_request(added_request_ids, internal=True)
556 raise e
557
558 return added_request_ids
559
560 def _add_request(
561 self,
562 prompt: EngineInput,
563 params: SamplingParams | PoolingParams,
564 lora_request: LoRARequest | None = None,
565 priority: int = 0,
566 ) -> str:
567 if isinstance(params, SamplingParams):
568 # We only care about the final output
569 params.output_kind = RequestOutputKind.FINAL_ONLY
570
571 request_id = str(next(self.request_counter))
572
573 return self.llm_engine.add_request(
574 request_id,
575 prompt,
576 params,
577 lora_request=lora_request,
578 priority=priority,
579 )
580
581 def _run_engine(
582 self,
583 output_type: type[_O] | tuple[type[_O], ...],
584 *,
585 use_tqdm: bool | Callable[..., tqdm] = True,
586 ) -> list[_O]:
587 # Initialize tqdm.
588 if use_tqdm:
589 num_requests = self.llm_engine.get_num_unfinished_requests()
590 tqdm_func = use_tqdm if callable(use_tqdm) else tqdm
591 pbar = tqdm_func(
592 total=num_requests,
593 desc="Processed prompts",
594 dynamic_ncols=True,
595 postfix=(f"est. speed input: {0:.2f} toks/s, output: {0:.2f} toks/s"),
596 )
597
598 # Run the engine.
599 outputs: list[_O] = []
600 total_in_toks = 0
601 total_out_toks = 0
602 while self.llm_engine.has_unfinished_requests():
603 step_outputs = self.llm_engine.step()
604 for output in step_outputs:
605 assert isinstance(output, output_type)
606 if output.finished:
607 outputs.append(output) # type: ignore[arg-type]
608 if use_tqdm:
609 if isinstance(output, RequestOutput):
610 # Calculate tokens only for RequestOutput
611 n = len(output.outputs)
612 assert output.prompt_token_ids is not None
613 total_in_toks += len(output.prompt_token_ids) * n
614 in_spd = total_in_toks / pbar.format_dict["elapsed"]
615 total_out_toks += sum(
616 len(stp.token_ids) for stp in output.outputs
617 )
618 out_spd = total_out_toks / pbar.format_dict["elapsed"]
619 pbar.postfix = (
620 f"est. speed input: {in_spd:.2f} toks/s, "
621 f"output: {out_spd:.2f} toks/s"
622 )
623 pbar.update(n)
624 else:
625 pbar.update(1)
626 if pbar.n == num_requests:
627 pbar.refresh()
628
629 if use_tqdm:
630 pbar.close()
631 # Sort the outputs by request ID.
632 # This is necessary because some requests may be finished earlier than
633 # its previous requests.
634 return sorted(outputs, key=lambda x: int(x.request_id))