| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import time |
| 5 | import weakref |
| 6 | from collections.abc import Callable, Mapping |
| 7 | from copy import copy |
| 8 | from typing import Any |
| 9 | |
| 10 | import torch.nn as nn |
| 11 | from typing_extensions import TypeVar |
| 12 | |
| 13 | import vllm.envs as envs |
| 14 | from vllm.config import ParallelConfig, VllmConfig |
| 15 | from vllm.distributed import stateless_destroy_torch_distributed_process_group |
| 16 | from vllm.distributed.parallel_state import get_dp_group |
| 17 | from vllm.engine.arg_utils import EngineArgs |
| 18 | from vllm.inputs import EngineInput, PromptType |
| 19 | from vllm.logger import configure_logging_if_needed, init_logger |
| 20 | from vllm.lora.request import LoRARequest |
| 21 | from vllm.multimodal import MULTIMODAL_REGISTRY, MultiModalRegistry |
| 22 | from vllm.outputs import PoolingRequestOutput, RequestOutput |
| 23 | from vllm.pooling_params import PoolingParams |
| 24 | from vllm.renderers import renderer_from_config |
| 25 | from vllm.renderers.inputs.preprocess import extract_prompt_components |
| 26 | from vllm.sampling_params import SamplingParams |
| 27 | from vllm.tasks import SupportedTask |
| 28 | from vllm.tokenizers import TokenizerLike |
| 29 | from vllm.tracing import init_tracer |
| 30 | from vllm.usage.usage_lib import UsageContext |
| 31 | from vllm.v1.engine import EngineCoreRequest, PauseMode |
| 32 | from vllm.v1.engine.core_client import EngineCoreClient |
| 33 | from vllm.v1.engine.input_processor import InputProcessor |
| 34 | from vllm.v1.engine.output_processor import OutputProcessor |
| 35 | from vllm.v1.engine.parallel_sampling import ParentRequest |
| 36 | from vllm.v1.executor import Executor |
| 37 | from vllm.v1.kv_hints import KvHintsEnvelope |
| 38 | from vllm.v1.metrics.loggers import StatLoggerFactory, StatLoggerManager |
| 39 | from vllm.v1.metrics.reader import Metric, get_metrics_snapshot |
| 40 | from vllm.v1.metrics.stats import IterationStats |
| 41 | from vllm.v1.utils import record_function_or_nullcontext |
| 42 | from vllm.v1.worker.worker_base import WorkerBase |
| 43 | |
| 44 | logger = init_logger(__name__) |
| 45 | |
| 46 | _R = TypeVar("_R", default=Any) |
| 47 | |
| 48 | |
| 49 | class LLMEngine: |
| 50 | """Legacy LLMEngine for backwards compatibility.""" |
| 51 | |
| 52 | def __init__( |
| 53 | self, |
| 54 | vllm_config: VllmConfig, |
| 55 | executor_class: type[Executor], |
| 56 | log_stats: bool, |
| 57 | aggregate_engine_logging: bool = False, |
| 58 | usage_context: UsageContext = UsageContext.ENGINE_CONTEXT, |
| 59 | stat_loggers: list[StatLoggerFactory] | None = None, |
| 60 | mm_registry: MultiModalRegistry = MULTIMODAL_REGISTRY, |
| 61 | multiprocess_mode: bool = False, |
| 62 | ) -> None: |
| 63 | configure_logging_if_needed(vllm_config.logging_config) |
| 64 | self.vllm_config = vllm_config |
| 65 | self.model_config = vllm_config.model_config |
| 66 | self.observability_config = vllm_config.observability_config |
| 67 | |
| 68 | tracing_endpoint = self.observability_config.otlp_traces_endpoint |
| 69 | if tracing_endpoint is not None: |
| 70 | init_tracer("vllm.llm_engine", tracing_endpoint) |
| 71 | |
| 72 | self.log_stats = log_stats |
| 73 | |
| 74 | parallel_config = vllm_config.parallel_config |
| 75 | executor_backend = parallel_config.distributed_executor_backend |
| 76 | |
| 77 | self.external_launcher_dp = ( |
| 78 | parallel_config.data_parallel_size > 1 |
| 79 | and executor_backend == "external_launcher" |
| 80 | ) |
| 81 | # important: init dp group before init the engine_core |
| 82 | # In the decoupled engine case this is handled in EngineCoreProc. |
| 83 | if ( |
| 84 | not multiprocess_mode |
| 85 | and parallel_config.data_parallel_size > 1 |
| 86 | and not self.external_launcher_dp |
| 87 | ): |
| 88 | self.dp_group = parallel_config.stateless_init_dp_group() |
| 89 | else: |
| 90 | self.dp_group = None |
| 91 | self.should_execute_dummy_batch = False |
| 92 | |
| 93 | self.renderer = renderer = renderer_from_config(self.vllm_config) |
| 94 | |
| 95 | # Convert EngineInput --> EngineCoreRequest. |
| 96 | self.input_processor = InputProcessor(self.vllm_config, renderer) |
| 97 | |
| 98 | # Converts EngineCoreOutputs --> RequestOutput. |
| 99 | self.output_processor = OutputProcessor( |
| 100 | renderer.tokenizer, |
| 101 | log_stats=self.log_stats, |
| 102 | stream_interval=self.vllm_config.scheduler_config.stream_interval, |
| 103 | tracing_enabled=tracing_endpoint is not None, |
| 104 | ) |
| 105 | |
| 106 | # EngineCore (gets EngineCoreRequests and gives EngineCoreOutputs) |
| 107 | # Hand the renderer to the client. In multiprocess mode the client |
| 108 | # starts the MM warmup only after engine-core fork (the why is in |
| 109 | # BaseRenderer.start_mm_warmup_in_background); InprocClient takes no |
| 110 | # renderer, so MM warmup stays inside renderer.warmup() there. |
| 111 | self.engine_core = EngineCoreClient.make_client( |
| 112 | multiprocess_mode=multiprocess_mode, |
| 113 | asyncio_mode=False, |
| 114 | vllm_config=vllm_config, |
| 115 | executor_class=executor_class, |
| 116 | log_stats=self.log_stats, |
| 117 | renderer=renderer, |
| 118 | ) |
| 119 | |
| 120 | self.logger_manager: StatLoggerManager | None = None |
| 121 | if self.log_stats: |
| 122 | self.logger_manager = StatLoggerManager( |
| 123 | vllm_config=vllm_config, |
| 124 | custom_stat_loggers=stat_loggers, |
| 125 | enable_default_loggers=log_stats, |
| 126 | aggregate_engine_logging=aggregate_engine_logging, |
| 127 | ) |
| 128 | self.logger_manager.log_engine_initialized() |
| 129 | |
| 130 | if not multiprocess_mode: |
| 131 | # for v0 compatibility |
| 132 | self.model_executor = self.engine_core.engine_core.model_executor # type: ignore |
| 133 | |
| 134 | # Capture the model while reachable so the finalizer can drop the |
| 135 | # bytecode hooks pinning it (frees GPU memory on engine deletion). |
| 136 | model = self._get_driver_model_for_cleanup() |
| 137 | if model is not None: |
| 138 | self._finalizer = weakref.finalize( |
| 139 | self, LLMEngine._cleanup_instance_caches, weakref.ref(model) |
| 140 | ) |
| 141 | |
| 142 | if self.external_launcher_dp: |
| 143 | # If we use DP in external launcher mode, we reuse the |
| 144 | # existing DP group used for data communication. |
| 145 | self.dp_group = get_dp_group().cpu_group |
| 146 | |
| 147 | # Don't keep the dummy data in memory |
| 148 | self.reset_mm_cache() |
| 149 | |
| 150 | @classmethod |
| 151 | def from_vllm_config( |
| 152 | cls, |
| 153 | vllm_config: VllmConfig, |
| 154 | usage_context: UsageContext = UsageContext.ENGINE_CONTEXT, |
| 155 | stat_loggers: list[StatLoggerFactory] | None = None, |
| 156 | disable_log_stats: bool = False, |
| 157 | ) -> "LLMEngine": |
| 158 | return cls( |
| 159 | vllm_config=vllm_config, |
| 160 | executor_class=Executor.get_class(vllm_config), |
| 161 | log_stats=(not disable_log_stats), |
| 162 | usage_context=usage_context, |
| 163 | stat_loggers=stat_loggers, |
| 164 | multiprocess_mode=envs.VLLM_ENABLE_V1_MULTIPROCESSING, |
| 165 | ) |
| 166 | |
| 167 | @classmethod |
| 168 | def from_engine_args( |
| 169 | cls, |
| 170 | engine_args: EngineArgs, |
| 171 | usage_context: UsageContext = UsageContext.ENGINE_CONTEXT, |
| 172 | stat_loggers: list[StatLoggerFactory] | None = None, |
| 173 | enable_multiprocessing: bool = False, |
| 174 | ) -> "LLMEngine": |
| 175 | """Creates an LLM engine from the engine arguments.""" |
| 176 | # Create the engine configs. |
| 177 | vllm_config = engine_args.create_engine_config(usage_context) |
| 178 | executor_class = Executor.get_class(vllm_config) |
| 179 | |
| 180 | if envs.VLLM_ENABLE_V1_MULTIPROCESSING: |
| 181 | logger.debug("Enabling multiprocessing for LLMEngine.") |
| 182 | enable_multiprocessing = True |
| 183 | |
| 184 | # Create the LLMEngine. |
| 185 | return cls( |
| 186 | vllm_config=vllm_config, |
| 187 | executor_class=executor_class, |
| 188 | log_stats=not engine_args.disable_log_stats, |
| 189 | usage_context=usage_context, |
| 190 | stat_loggers=stat_loggers, |
| 191 | multiprocess_mode=enable_multiprocessing, |
| 192 | ) |
| 193 | |
| 194 | def get_num_unfinished_requests(self) -> int: |
| 195 | return self.output_processor.get_num_unfinished_requests() |
| 196 | |
| 197 | def has_unfinished_requests(self) -> bool: |
| 198 | has_unfinished = self.output_processor.has_unfinished_requests() |
| 199 | if self.dp_group is None: |
| 200 | return has_unfinished or self.engine_core.dp_engines_running() |
| 201 | return self.has_unfinished_requests_dp(has_unfinished) |
| 202 | |
| 203 | def has_unfinished_requests_dp(self, has_unfinished: bool) -> bool: |
| 204 | aggregated_has_unfinished = ParallelConfig.has_unfinished_dp( |
| 205 | self.dp_group, has_unfinished |
| 206 | ) |
| 207 | if not has_unfinished and aggregated_has_unfinished: |
| 208 | self.should_execute_dummy_batch = True |
| 209 | return aggregated_has_unfinished |
| 210 | |
| 211 | def get_supported_tasks(self) -> tuple[SupportedTask, ...]: |
| 212 | if not hasattr(self, "_supported_tasks"): |
| 213 | # Cache the result |
| 214 | self._supported_tasks = self.engine_core.get_supported_tasks() |
| 215 | |
| 216 | return self._supported_tasks |
| 217 | |
| 218 | def abort_request(self, request_ids: list[str], internal: bool = False) -> None: |
| 219 | """Remove request_ids from EngineCore and Detokenizer.""" |
| 220 | request_ids = self.output_processor.abort_requests(request_ids, internal) |
| 221 | self.engine_core.abort_requests(request_ids) |
| 222 | |
| 223 | def add_request( |
| 224 | self, |
| 225 | request_id: str, |
| 226 | prompt: EngineCoreRequest | PromptType | EngineInput, |
| 227 | params: SamplingParams | PoolingParams, |
| 228 | arrival_time: float | None = None, |
| 229 | lora_request: LoRARequest | None = None, |
| 230 | tokenization_kwargs: dict[str, Any] | None = None, |
| 231 | trace_headers: Mapping[str, str] | None = None, |
| 232 | priority: int = 0, |
| 233 | session_id: str | None = None, |
| 234 | prompt_text: str | None = None, |
| 235 | kv_hints: KvHintsEnvelope | None = None, |
| 236 | ) -> str: |
| 237 | # Validate the request_id type. |
| 238 | if not isinstance(request_id, str): |
| 239 | raise TypeError(f"request_id must be a string, got {type(request_id)}") |
| 240 | |
| 241 | # Process raw inputs into the request. |
| 242 | if isinstance(prompt, EngineCoreRequest): |
| 243 | logger.warning_once( |
| 244 | "Passing EngineCoreRequest to LLMEngine.generate() and .add_requests() " |
| 245 | "is deprecated and will be removed in the future. You should " |
| 246 | "instead pass the outputs of Renderer.render_cmpl() or " |
| 247 | "Renderer.render_chat()." |
| 248 | ) |
| 249 | |
| 250 | request = prompt |
| 251 | if request_id != request.request_id: |
| 252 | logger.warning_once( |
| 253 | "LLMEngine.add_request() was passed a request_id parameter that " |
| 254 | "does not match the EngineCoreRequest.request_id attribute. The " |
| 255 | "latter will be used, and the former will be ignored." |
| 256 | ) |
| 257 | else: |
| 258 | request = self.input_processor.process_inputs( |
| 259 | request_id, |
| 260 | prompt, |
| 261 | params, |
| 262 | supported_tasks=self.get_supported_tasks(), |
| 263 | arrival_time=arrival_time, |
| 264 | lora_request=lora_request, |
| 265 | tokenization_kwargs=tokenization_kwargs, |
| 266 | trace_headers=trace_headers, |
| 267 | priority=priority, |
| 268 | session_id=session_id, |
| 269 | kv_hints=kv_hints, |
| 270 | ) |
| 271 | prompt_text, _, _ = extract_prompt_components(self.model_config, prompt) |
| 272 | |
| 273 | self.input_processor.assign_request_id(request) |
| 274 | |
| 275 | req_id = request.request_id |
| 276 | |
| 277 | # Use cloned params that may have been updated in process_inputs() |
| 278 | params = request.params |
| 279 | |
| 280 | n = params.n if isinstance(params, SamplingParams) else 1 |
| 281 | |
| 282 | if n == 1: |
| 283 | # Make a new RequestState and queue. |
| 284 | self.output_processor.add_request(request, prompt_text, None, 0) |
| 285 | # Add the request to EngineCore. |
| 286 | self.engine_core.add_request(request) |
| 287 | return req_id |
| 288 | |
| 289 | # Fan out child requests (for n>1). |
| 290 | parent_req = ParentRequest(request) |
| 291 | for idx in range(n): |
| 292 | request_id, child_params = parent_req.get_child_info(idx) |
| 293 | child_request = request if idx == n - 1 else copy(request) |
| 294 | child_request.request_id = request_id |
| 295 | child_request.sampling_params = child_params |
| 296 | |
| 297 | # Make a new RequestState and queue. |
| 298 | self.output_processor.add_request( |
| 299 | child_request, prompt_text, parent_req, idx |
| 300 | ) |
| 301 | # Add the request to EngineCore. |
| 302 | self.engine_core.add_request(child_request) |
| 303 | |
| 304 | return req_id |
| 305 | |
| 306 | def step(self) -> list[RequestOutput | PoolingRequestOutput]: |
| 307 | if self.should_execute_dummy_batch: |
| 308 | self.should_execute_dummy_batch = False |
| 309 | self.engine_core.execute_dummy_batch() |
| 310 | return [] |
| 311 | |
| 312 | # 1) Get EngineCoreOutput from the EngineCore. |
| 313 | with record_function_or_nullcontext("llm_engine step: get_output"): |
| 314 | outputs = self.engine_core.get_output() |
| 315 | |
| 316 | # 2) Process EngineCoreOutputs. |
| 317 | with record_function_or_nullcontext("llm_engine step: process_outputs"): |
| 318 | iteration_stats = ( |
| 319 | IterationStats() if self.log_stats and outputs.outputs else None |
| 320 | ) |
| 321 | processed_outputs = self.output_processor.process_outputs( |
| 322 | outputs.outputs, |
| 323 | engine_core_timestamp=outputs.timestamp, |
| 324 | iteration_stats=iteration_stats, |
| 325 | ) |
| 326 | |
| 327 | mm_processor_cache = self.renderer.mm_processor_cache |
| 328 | if mm_processor_cache is not None: |
| 329 | for engine_core_output in outputs.outputs: |
| 330 | if engine_core_output.mm_cache_miss_hashes: |
| 331 | for mm_hash in engine_core_output.mm_cache_miss_hashes: |
| 332 | mm_processor_cache.invalidate(mm_hash) |
| 333 | |
| 334 | self.output_processor.update_scheduler_stats(outputs.scheduler_stats) |
| 335 | |
| 336 | # 3) Abort any reqs that finished due to stop strings. |
| 337 | with record_function_or_nullcontext("llm_engine step: abort_requests"): |
| 338 | self.engine_core.abort_requests(processed_outputs.reqs_to_abort) |
| 339 | |
| 340 | # 4) Record stats |
| 341 | with record_function_or_nullcontext("llm_engine step: record_stats"): |
| 342 | if self.logger_manager is not None and outputs.scheduler_stats is not None: |
| 343 | # Record even when this step produced no request outputs. |
| 344 | self.logger_manager.record( |
| 345 | scheduler_stats=outputs.scheduler_stats, |
| 346 | iteration_stats=iteration_stats, |
| 347 | mm_cache_stats=self.renderer.stat_mm_cache(), |
| 348 | ) |
| 349 | if outputs.outputs: |
| 350 | self.do_log_stats_with_interval() |
| 351 | |
| 352 | return processed_outputs.request_outputs |
| 353 | |
| 354 | def start_profile(self, profile_prefix: str | None = None): |
| 355 | self.engine_core.profile(True, profile_prefix) |
| 356 | |
| 357 | def stop_profile(self): |
| 358 | self.engine_core.profile(False) |
| 359 | |
| 360 | def reset_mm_cache(self): |
| 361 | # Join the background MM warmup first: the mm_processor_cache is not |
| 362 | # safe for concurrent access with its apply/clear. |
| 363 | self.renderer._join_mm_warmup() |
| 364 | self.renderer.clear_mm_cache() |
| 365 | self.engine_core.reset_mm_cache() |
| 366 | |
| 367 | def reset_prefix_cache( |
| 368 | self, reset_running_requests: bool = False, reset_connector: bool = False |
| 369 | ) -> bool: |
| 370 | return self.engine_core.reset_prefix_cache( |
| 371 | reset_running_requests, reset_connector |
| 372 | ) |
| 373 | |
| 374 | def reset_encoder_cache(self) -> None: |
| 375 | """Reset the encoder cache to invalidate all cached encoder outputs. |
| 376 | |
| 377 | This should be called when model weights are updated to ensure |
| 378 | stale vision embeddings computed with old weights are not reused. |
| 379 | """ |
| 380 | self.engine_core.reset_encoder_cache() |
| 381 | |
| 382 | def sleep(self, level: int = 1, mode: PauseMode = "abort"): |
| 383 | if level >= 1: |
| 384 | self.renderer.clear_mm_cache() |
| 385 | self.engine_core.sleep(level, mode) |
| 386 | |
| 387 | if self.logger_manager is not None: |
| 388 | self.logger_manager.record_sleep_state(1, level) |
| 389 | |
| 390 | def release_kv_cache_memory(self) -> None: |
| 391 | self.renderer.clear_mm_cache() |
| 392 | self.engine_core.release_kv_cache_memory() |
| 393 | |
| 394 | if self.logger_manager is not None: |
| 395 | self.logger_manager.record_sleep_state(1, 0) |
| 396 | |
| 397 | def wake_up(self, tags: list[str] | None = None) -> bool: |
| 398 | fully_awake = self.engine_core.wake_up(tags) |
| 399 | |
| 400 | if self.logger_manager is not None and fully_awake: |
| 401 | self.logger_manager.record_sleep_state(0, 0) |
| 402 | |
| 403 | return fully_awake |
| 404 | |
| 405 | def is_sleeping(self) -> bool: |
| 406 | return self.engine_core.is_sleeping() |
| 407 | |
| 408 | def get_metrics(self) -> list[Metric]: |
| 409 | assert self.log_stats, "Stat logging disabled" |
| 410 | return get_metrics_snapshot() |
| 411 | |
| 412 | @property |
| 413 | def tokenizer(self) -> TokenizerLike | None: |
| 414 | return self.renderer.tokenizer |
| 415 | |
| 416 | def get_tokenizer(self) -> TokenizerLike: |
| 417 | return self.renderer.get_tokenizer() |
| 418 | |
| 419 | def do_log_stats(self) -> None: |
| 420 | """Log stats if logging is enabled.""" |
| 421 | if self.logger_manager: |
| 422 | self.logger_manager.log() |
| 423 | |
| 424 | def do_log_stats_with_interval(self) -> None: |
| 425 | """Log stats when the time interval has passed.""" |
| 426 | now = time.time() |
| 427 | if not hasattr(self, "_last_log_time"): |
| 428 | self._last_log_time = now |
| 429 | if now - self._last_log_time >= envs.VLLM_LOG_STATS_INTERVAL: |
| 430 | self.do_log_stats() |
| 431 | self._last_log_time = now |
| 432 | |
| 433 | def add_lora(self, lora_request: LoRARequest) -> bool: |
| 434 | """Load a new LoRA adapter into the engine for future requests.""" |
| 435 | return self.engine_core.add_lora(lora_request) |
| 436 | |
| 437 | def remove_lora(self, lora_id: int) -> bool: |
| 438 | """Remove an already loaded LoRA adapter.""" |
| 439 | return self.engine_core.remove_lora(lora_id) |
| 440 | |
| 441 | def list_loras(self) -> set[int]: |
| 442 | """List all registered adapters.""" |
| 443 | return self.engine_core.list_loras() |
| 444 | |
| 445 | def pin_lora(self, lora_id: int) -> bool: |
| 446 | """Prevent an adapter from being evicted.""" |
| 447 | return self.engine_core.pin_lora(lora_id) |
| 448 | |
| 449 | def collective_rpc( |
| 450 | self, |
| 451 | method: str | Callable[[WorkerBase], _R], |
| 452 | timeout: float | None = None, |
| 453 | args: tuple = (), |
| 454 | kwargs: dict[str, Any] | None = None, |
| 455 | ) -> list[_R]: |
| 456 | return self.engine_core.collective_rpc(method, timeout, args, kwargs) |
| 457 | |
| 458 | def set_weight_version(self, weight_version: str) -> None: |
| 459 | self.engine_core.set_weight_version(weight_version) |
| 460 | |
| 461 | def get_weight_version(self) -> str: |
| 462 | """Return the latest committed weight version.""" |
| 463 | return self.engine_core.get_weight_version() |
| 464 | |
| 465 | def apply_model(self, func: Callable[[nn.Module], _R]) -> list[_R]: |
| 466 | return self.collective_rpc("apply_model", args=(func,)) |
| 467 | |
| 468 | def _get_driver_model_for_cleanup(self) -> nn.Module | None: |
| 469 | driver_worker = getattr(self.model_executor, "driver_worker", None) |
| 470 | model_runner = getattr(driver_worker, "model_runner", None) |
| 471 | return getattr(model_runner, "model", None) |
| 472 | |
| 473 | @staticmethod |
| 474 | def _cleanup_instance_caches(model_ref: "weakref.ref[nn.Module]") -> None: |
| 475 | """Remove the bytecode hooks that pin the compiled model.""" |
| 476 | from vllm.compilation.wrapper import TorchCompileWithNoGuardsWrapper |
| 477 | |
| 478 | model = model_ref() |
| 479 | if model is None: |
| 480 | return |
| 481 | for module in model.modules(): |
| 482 | if isinstance(module, TorchCompileWithNoGuardsWrapper): |
| 483 | module.cleanup() |
| 484 | |
| 485 | def __del__(self): |
| 486 | dp_group = getattr(self, "dp_group", None) |
| 487 | if dp_group is not None and not self.external_launcher_dp: |
| 488 | stateless_destroy_torch_distributed_process_group(dp_group) |