llm_engine.py 16.1 KB
Newer Older
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 copy import copy
7
from typing import Any
8

9
import torch.nn as nn
10
from typing_extensions import TypeVar
11

12
import vllm.envs as envs
13
from vllm.config import ParallelConfig, VllmConfig
14
from vllm.distributed import stateless_destroy_torch_distributed_process_group
15
from vllm.distributed.parallel_state import get_dp_group
16
from vllm.engine.arg_utils import EngineArgs
17
from vllm.inputs import EngineInput, PromptType
18
19
from vllm.logger import init_logger
from vllm.lora.request import LoRARequest
20
from vllm.multimodal import MULTIMODAL_REGISTRY, MultiModalRegistry
21
from vllm.outputs import PoolingRequestOutput, RequestOutput
22
from vllm.pooling_params import PoolingParams
23
from vllm.renderers import renderer_from_config
24
from vllm.renderers.inputs.preprocess import extract_prompt_components
25
from vllm.sampling_params import SamplingParams
26
from vllm.tasks import SupportedTask
27
from vllm.tokenizers import TokenizerLike
28
from vllm.tracing import init_tracer
29
from vllm.usage.usage_lib import UsageContext
30
from vllm.v1.engine import EngineCoreRequest, PauseMode
31
from vllm.v1.engine.core_client import EngineCoreClient
32
from vllm.v1.engine.input_processor import InputProcessor
33
from vllm.v1.engine.output_processor import OutputProcessor
34
from vllm.v1.engine.parallel_sampling import ParentRequest
35
from vllm.v1.executor import Executor
36
from vllm.v1.metrics.loggers import StatLoggerFactory, StatLoggerManager
37
38
from vllm.v1.metrics.reader import Metric, get_metrics_snapshot
from vllm.v1.metrics.stats import IterationStats
39
from vllm.v1.utils import record_function_or_nullcontext
40
from vllm.v1.worker.worker_base import WorkerBase
41
42
43

logger = init_logger(__name__)

44
_R = TypeVar("_R", default=Any)
45

46
47

class LLMEngine:
48
    """Legacy LLMEngine for backwards compatibility."""
49
50
51

    def __init__(
        self,
52
        vllm_config: VllmConfig,
53
        executor_class: type[Executor],
54
        log_stats: bool,
55
        aggregate_engine_logging: bool = False,
56
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
57
        stat_loggers: list[StatLoggerFactory] | None = None,
58
        mm_registry: MultiModalRegistry = MULTIMODAL_REGISTRY,
59
        multiprocess_mode: bool = False,
60
    ) -> None:
61
        self.vllm_config = vllm_config
62
        self.model_config = vllm_config.model_config
63
64
65
66
67
        self.observability_config = vllm_config.observability_config

        tracing_endpoint = self.observability_config.otlp_traces_endpoint
        if tracing_endpoint is not None:
            init_tracer("vllm.llm_engine", tracing_endpoint)
68

69
70
        self.log_stats = log_stats

71
        parallel_config = vllm_config.parallel_config
72
73
        executor_backend = parallel_config.distributed_executor_backend

74
75
76
77
        self.external_launcher_dp = (
            parallel_config.data_parallel_size > 1
            and executor_backend == "external_launcher"
        )
78
        # important: init dp group before init the engine_core
79
        # In the decoupled engine case this is handled in EngineCoreProc.
80
81
82
83
84
        if (
            not multiprocess_mode
            and parallel_config.data_parallel_size > 1
            and not self.external_launcher_dp
        ):
85
86
87
            self.dp_group = parallel_config.stateless_init_dp_group()
        else:
            self.dp_group = None
88
89
        self.should_execute_dummy_batch = False

90
        self.renderer = renderer = renderer_from_config(self.vllm_config)
91

92
        # Convert EngineInput --> EngineCoreRequest.
93
94
95
        self.input_processor = InputProcessor(self.vllm_config, renderer)

        # Converts EngineCoreOutputs --> RequestOutput.
96
        self.output_processor = OutputProcessor(
97
            renderer.tokenizer,
98
99
            log_stats=self.log_stats,
            stream_interval=self.vllm_config.scheduler_config.stream_interval,
100
            tracing_enabled=tracing_endpoint is not None,
101
        )
102
103
104
105
106

        # EngineCore (gets EngineCoreRequests and gives EngineCoreOutputs)
        self.engine_core = EngineCoreClient.make_client(
            multiprocess_mode=multiprocess_mode,
            asyncio_mode=False,
107
108
            vllm_config=vllm_config,
            executor_class=executor_class,
109
            log_stats=self.log_stats,
110
        )
111

112
        self.logger_manager: StatLoggerManager | None = None
113
114
115
116
117
        if self.log_stats:
            self.logger_manager = StatLoggerManager(
                vllm_config=vllm_config,
                custom_stat_loggers=stat_loggers,
                enable_default_loggers=log_stats,
118
                aggregate_engine_logging=aggregate_engine_logging,
119
120
121
            )
            self.logger_manager.log_engine_initialized()

122
123
124
125
        if not multiprocess_mode:
            # for v0 compatibility
            self.model_executor = self.engine_core.engine_core.model_executor  # type: ignore

126
127
128
129
130
        if self.external_launcher_dp:
            # If we use DP in external launcher mode, we reuse the
            # existing DP group used for data communication.
            self.dp_group = get_dp_group().cpu_group

131
132
133
        # Don't keep the dummy data in memory
        self.reset_mm_cache()

134
135
136
137
138
    @classmethod
    def from_vllm_config(
        cls,
        vllm_config: VllmConfig,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
139
        stat_loggers: list[StatLoggerFactory] | None = None,
140
141
        disable_log_stats: bool = False,
    ) -> "LLMEngine":
142
143
144
145
146
147
148
149
        return cls(
            vllm_config=vllm_config,
            executor_class=Executor.get_class(vllm_config),
            log_stats=(not disable_log_stats),
            usage_context=usage_context,
            stat_loggers=stat_loggers,
            multiprocess_mode=envs.VLLM_ENABLE_V1_MULTIPROCESSING,
        )
150

151
152
153
154
155
    @classmethod
    def from_engine_args(
        cls,
        engine_args: EngineArgs,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
156
        stat_loggers: list[StatLoggerFactory] | None = None,
157
        enable_multiprocessing: bool = False,
158
159
    ) -> "LLMEngine":
        """Creates an LLM engine from the engine arguments."""
160

161
        # Create the engine configs.
162
        vllm_config = engine_args.create_engine_config(usage_context)
163
        executor_class = Executor.get_class(vllm_config)
164

165
        if envs.VLLM_ENABLE_V1_MULTIPROCESSING:
166
167
168
169
            logger.debug("Enabling multiprocessing for LLMEngine.")
            enable_multiprocessing = True

        # Create the LLMEngine.
170
171
172
173
174
175
176
177
        return cls(
            vllm_config=vllm_config,
            executor_class=executor_class,
            log_stats=not engine_args.disable_log_stats,
            usage_context=usage_context,
            stat_loggers=stat_loggers,
            multiprocess_mode=enable_multiprocessing,
        )
178
179

    def get_num_unfinished_requests(self) -> int:
180
        return self.output_processor.get_num_unfinished_requests()
181
182

    def has_unfinished_requests(self) -> bool:
183
        has_unfinished = self.output_processor.has_unfinished_requests()
184
        if self.dp_group is None:
185
            return has_unfinished or self.engine_core.dp_engines_running()
186
187
188
189
        return self.has_unfinished_requests_dp(has_unfinished)

    def has_unfinished_requests_dp(self, has_unfinished: bool) -> bool:
        aggregated_has_unfinished = ParallelConfig.has_unfinished_dp(
190
191
            self.dp_group, has_unfinished
        )
192
193
194
        if not has_unfinished and aggregated_has_unfinished:
            self.should_execute_dummy_batch = True
        return aggregated_has_unfinished
195

196
    def get_supported_tasks(self) -> tuple[SupportedTask, ...]:
197
198
199
200
201
        if not hasattr(self, "_supported_tasks"):
            # Cache the result
            self._supported_tasks = self.engine_core.get_supported_tasks()

        return self._supported_tasks
202

203
    def abort_request(self, request_ids: list[str], internal: bool = False) -> None:
204
205
        """Remove request_ids from EngineCore and Detokenizer."""

206
        request_ids = self.output_processor.abort_requests(request_ids, internal)
207
208
        self.engine_core.abort_requests(request_ids)

209
210
211
    def add_request(
        self,
        request_id: str,
212
        prompt: EngineCoreRequest | PromptType | EngineInput,
213
214
215
216
217
        params: SamplingParams | PoolingParams,
        arrival_time: float | None = None,
        lora_request: LoRARequest | None = None,
        tokenization_kwargs: dict[str, Any] | None = None,
        trace_headers: Mapping[str, str] | None = None,
218
        priority: int = 0,
219
        prompt_text: str | None = None,
220
    ) -> str:
221
222
        # Validate the request_id type.
        if not isinstance(request_id, str):
223
            raise TypeError(f"request_id must be a string, got {type(request_id)}")
224

225
        # Process raw inputs into the request.
226
        if isinstance(prompt, EngineCoreRequest):
227
228
229
230
231
232
            logger.warning_once(
                "Passing EngineCoreRequest to LLMEngine.generate() and .add_requests() "
                "is deprecated and will be removed in v0.18. You should instead pass "
                "the outputs of Renderer.render_cmpl() or Renderer.render_chat()."
            )

233
            request = prompt
234
235
            if request_id != request.request_id:
                logger.warning_once(
236
                    "LLMEngine.add_request() was passed a request_id parameter that "
237
238
239
                    "does not match the EngineCoreRequest.request_id attribute. The "
                    "latter will be used, and the former will be ignored."
                )
240
        else:
241
            request = self.input_processor.process_inputs(
242
243
244
                request_id,
                prompt,
                params,
245
                supported_tasks=self.get_supported_tasks(),
246
247
248
249
250
                arrival_time=arrival_time,
                lora_request=lora_request,
                tokenization_kwargs=tokenization_kwargs,
                trace_headers=trace_headers,
                priority=priority,
251
            )
252
            prompt_text, _, _ = extract_prompt_components(self.model_config, prompt)
253

254
255
        self.input_processor.assign_request_id(request)

256
257
        req_id = request.request_id

258
259
260
        # Use cloned params that may have been updated in process_inputs()
        params = request.params

261
        n = params.n if isinstance(params, SamplingParams) else 1
262

263
264
        if n == 1:
            # Make a new RequestState and queue.
265
            self.output_processor.add_request(request, prompt_text, None, 0)
266
            # Add the request to EngineCore.
267
            self.engine_core.add_request(request)
268
            return req_id
269
270

        # Fan out child requests (for n>1).
271
        parent_req = ParentRequest(request)
272
        for idx in range(n):
273
            request_id, child_params = parent_req.get_child_info(idx)
274
275
            child_request = request if idx == n - 1 else copy(request)
            child_request.request_id = request_id
276
            child_request.sampling_params = child_params
277
278

            # Make a new RequestState and queue.
279
280
281
            self.output_processor.add_request(
                child_request, prompt_text, parent_req, idx
            )
282
283
            # Add the request to EngineCore.
            self.engine_core.add_request(child_request)
284

285
286
        return req_id

287
    def step(self) -> list[RequestOutput | PoolingRequestOutput]:
288
289
290
291
292
        if self.should_execute_dummy_batch:
            self.should_execute_dummy_batch = False
            self.engine_core.execute_dummy_batch()
            return []

293
        # 1) Get EngineCoreOutput from the EngineCore.
294
        with record_function_or_nullcontext("llm_engine step: get_output"):
295
            outputs = self.engine_core.get_output()
296

297
        # 2) Process EngineCoreOutputs.
298
        with record_function_or_nullcontext("llm_engine step: process_outputs"):
299
300
301
302
303
304
305
            iteration_stats = IterationStats() if self.log_stats else None
            processed_outputs = self.output_processor.process_outputs(
                outputs.outputs,
                engine_core_timestamp=outputs.timestamp,
                iteration_stats=iteration_stats,
            )
            self.output_processor.update_scheduler_stats(outputs.scheduler_stats)
306

307
        # 3) Abort any reqs that finished due to stop strings.
308
        with record_function_or_nullcontext("llm_engine step: abort_requests"):
309
            self.engine_core.abort_requests(processed_outputs.reqs_to_abort)
310

311
        # 4) Record stats
312
        with record_function_or_nullcontext("llm_engine step: record_stats"):
313
314
315
316
317
            if (
                self.logger_manager is not None
                and outputs.scheduler_stats is not None
                and len(outputs.outputs) > 0
            ):
318
319
320
                self.logger_manager.record(
                    scheduler_stats=outputs.scheduler_stats,
                    iteration_stats=iteration_stats,
321
                    mm_cache_stats=self.renderer.stat_mm_cache(),
322
323
                )
                self.do_log_stats_with_interval()
324

325
        return processed_outputs.request_outputs
326

327
328
    def start_profile(self, profile_prefix: str | None = None):
        self.engine_core.profile(True, profile_prefix)
329

330
    def stop_profile(self):
331
        self.engine_core.profile(False)
332

333
    def reset_mm_cache(self):
334
        self.renderer.clear_mm_cache()
335
336
        self.engine_core.reset_mm_cache()

337
338
339
340
341
342
    def reset_prefix_cache(
        self, reset_running_requests: bool = False, reset_connector: bool = False
    ) -> bool:
        return self.engine_core.reset_prefix_cache(
            reset_running_requests, reset_connector
        )
343

344
345
346
347
348
349
350
351
    def reset_encoder_cache(self) -> None:
        """Reset the encoder cache to invalidate all cached encoder outputs.

        This should be called when model weights are updated to ensure
        stale vision embeddings computed with old weights are not reused.
        """
        self.engine_core.reset_encoder_cache()

352
353
    def sleep(self, level: int = 1, mode: PauseMode = "abort"):
        self.engine_core.sleep(level, mode)
354

355
356
357
        if self.logger_manager is not None:
            self.logger_manager.record_sleep_state(1, level)

358
    def wake_up(self, tags: list[str] | None = None):
359
        self.engine_core.wake_up(tags)
360

361
362
363
        if self.logger_manager is not None:
            self.logger_manager.record_sleep_state(0, 0)

364
365
366
    def is_sleeping(self) -> bool:
        return self.engine_core.is_sleeping()

367
368
369
370
    def get_metrics(self) -> list[Metric]:
        assert self.log_stats, "Stat logging disabled"
        return get_metrics_snapshot()

371
    @property
372
    def tokenizer(self) -> TokenizerLike | None:
373
        return self.renderer.tokenizer
374

375
    def get_tokenizer(self) -> TokenizerLike:
376
        return self.renderer.get_tokenizer()
377

378
379
380
381
382
383
384
385
386
387
388
389
390
391
    def do_log_stats(self) -> None:
        """Log stats if logging is enabled."""
        if self.logger_manager:
            self.logger_manager.log()

    def do_log_stats_with_interval(self) -> None:
        """Log stats when the time interval has passed."""
        now = time.time()
        if not hasattr(self, "_last_log_time"):
            self._last_log_time = now
        if now - self._last_log_time >= envs.VLLM_LOG_STATS_INTERVAL:
            self.do_log_stats()
            self._last_log_time = now

392
393
394
395
396
397
398
399
    def add_lora(self, lora_request: LoRARequest) -> bool:
        """Load a new LoRA adapter into the engine for future requests."""
        return self.engine_core.add_lora(lora_request)

    def remove_lora(self, lora_id: int) -> bool:
        """Remove an already loaded LoRA adapter."""
        return self.engine_core.remove_lora(lora_id)

400
    def list_loras(self) -> set[int]:
401
402
403
404
405
406
        """List all registered adapters."""
        return self.engine_core.list_loras()

    def pin_lora(self, lora_id: int) -> bool:
        """Prevent an adapter from being evicted."""
        return self.engine_core.pin_lora(lora_id)
407

408
409
    def collective_rpc(
        self,
410
411
        method: str | Callable[[WorkerBase], _R],
        timeout: float | None = None,
412
        args: tuple = (),
413
        kwargs: dict[str, Any] | None = None,
414
    ) -> list[_R]:
415
416
        return self.engine_core.collective_rpc(method, timeout, args, kwargs)

417
    def apply_model(self, func: Callable[[nn.Module], _R]) -> list[_R]:
418
        return self.collective_rpc("apply_model", args=(func,))
419

420
    def __del__(self):
421
422
        dp_group = getattr(self, "dp_group", None)
        if dp_group is not None and not self.external_launcher_dp:
423
            stateless_destroy_torch_distributed_process_group(dp_group)