llm_engine.py 15.5 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, cast
8

9
import torch.nn as nn
10
from typing_extensions import TypeVar, deprecated
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 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.plugins.io_processors import get_io_processor
23
from vllm.pooling_params import PoolingParams
24
from vllm.sampling_params import SamplingParams
25
from vllm.tasks import SupportedTask
26
from vllm.tokenizers import TokenizerLike, init_tokenizer_from_config
27
from vllm.tracing import init_tracer
28
from vllm.usage.usage_lib import UsageContext
29
from vllm.v1.engine import EngineCoreRequest
30
from vllm.v1.engine.core_client import EngineCoreClient
31
from vllm.v1.engine.input_processor import InputProcessor
32
from vllm.v1.engine.output_processor import OutputProcessor
33
from vllm.v1.engine.parallel_sampling import ParentRequest
34
from vllm.v1.executor import Executor
35
from vllm.v1.metrics.loggers import StatLoggerFactory, StatLoggerManager
36
37
from vllm.v1.metrics.reader import Metric, get_metrics_snapshot
from vllm.v1.metrics.stats import IterationStats
38
from vllm.v1.utils import record_function_or_nullcontext
39
from vllm.v1.worker.worker_base import WorkerBase
40
41
42

logger = init_logger(__name__)

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

45
46

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

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

66
67
        self.log_stats = log_stats

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

86
87
88
        if self.model_config.skip_tokenizer_init:
            tokenizer = None
        else:
89
            tokenizer = init_tokenizer_from_config(self.model_config)
90

91
        self.input_processor = InputProcessor(self.vllm_config, tokenizer)
92
93
94
95
        self.io_processor = get_io_processor(
            self.vllm_config,
            self.model_config.io_processor_plugin,
        )
96

97
        # OutputProcessor (convert EngineCoreOutputs --> RequestOutput).
98
        self.output_processor = OutputProcessor(
99
100
101
            self.tokenizer,
            log_stats=self.log_stats,
            stream_interval=self.vllm_config.scheduler_config.stream_interval,
102
        )
103
104
105
        endpoint = self.observability_config.otlp_traces_endpoint
        if endpoint is not None:
            tracer = init_tracer("vllm.llm_engine", endpoint)
106
            self.output_processor.tracer = tracer
107
108
109
110
111

        # EngineCore (gets EngineCoreRequests and gives EngineCoreOutputs)
        self.engine_core = EngineCoreClient.make_client(
            multiprocess_mode=multiprocess_mode,
            asyncio_mode=False,
112
113
            vllm_config=vllm_config,
            executor_class=executor_class,
114
            log_stats=self.log_stats,
115
        )
116

117
        self.logger_manager: StatLoggerManager | None = None
118
119
120
121
122
        if self.log_stats:
            self.logger_manager = StatLoggerManager(
                vllm_config=vllm_config,
                custom_stat_loggers=stat_loggers,
                enable_default_loggers=log_stats,
123
                aggregate_engine_logging=aggregate_engine_logging,
124
125
126
            )
            self.logger_manager.log_engine_initialized()

127
128
129
130
        if not multiprocess_mode:
            # for v0 compatibility
            self.model_executor = self.engine_core.engine_core.model_executor  # type: ignore

131
132
133
134
135
        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

136
137
138
        # Don't keep the dummy data in memory
        self.reset_mm_cache()

139
140
141
142
143
144
145
146
    @property
    @deprecated(
        "`LLMEngine.processor` has been renamed to `LLMEngine.input_processor`. "
        "The old name will be removed in v0.13."
    )
    def processor(self):
        return self.input_processor

147
148
149
150
151
    @classmethod
    def from_vllm_config(
        cls,
        vllm_config: VllmConfig,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
152
        stat_loggers: list[StatLoggerFactory] | None = None,
153
154
        disable_log_stats: bool = False,
    ) -> "LLMEngine":
155
156
157
158
159
160
161
162
        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,
        )
163

164
165
166
167
168
    @classmethod
    def from_engine_args(
        cls,
        engine_args: EngineArgs,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
169
        stat_loggers: list[StatLoggerFactory] | None = None,
170
        enable_multiprocessing: bool = False,
171
172
    ) -> "LLMEngine":
        """Creates an LLM engine from the engine arguments."""
173

174
        # Create the engine configs.
175
        vllm_config = engine_args.create_engine_config(usage_context)
176
        executor_class = Executor.get_class(vllm_config)
177

178
        if envs.VLLM_ENABLE_V1_MULTIPROCESSING:
179
180
181
182
            logger.debug("Enabling multiprocessing for LLMEngine.")
            enable_multiprocessing = True

        # Create the LLMEngine.
183
184
185
186
187
188
189
190
        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,
        )
191
192

    def get_num_unfinished_requests(self) -> int:
193
        return self.output_processor.get_num_unfinished_requests()
194
195

    def has_unfinished_requests(self) -> bool:
196
        has_unfinished = self.output_processor.has_unfinished_requests()
197
        if self.dp_group is None:
198
            return has_unfinished or self.engine_core.dp_engines_running()
199
200
201
202
        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(
203
204
            self.dp_group, has_unfinished
        )
205
206
207
        if not has_unfinished and aggregated_has_unfinished:
            self.should_execute_dummy_batch = True
        return aggregated_has_unfinished
208
209
210
211
212

    @classmethod
    def validate_outputs(cls, outputs, output_type):
        return outputs

213
214
215
    def get_supported_tasks(self) -> tuple[SupportedTask, ...]:
        return self.engine_core.get_supported_tasks()

216
    def abort_request(self, request_ids: list[str]) -> None:
217
218
        """Remove request_ids from EngineCore and Detokenizer."""

219
        request_ids = self.output_processor.abort_requests(request_ids)
220
221
        self.engine_core.abort_requests(request_ids)

222
223
224
    def add_request(
        self,
        request_id: str,
225
226
227
228
229
230
        prompt: EngineCoreRequest | PromptType,
        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,
231
        priority: int = 0,
232
        prompt_text: str | None = None,
233
    ) -> None:
234
235
        # Validate the request_id type.
        if not isinstance(request_id, str):
236
            raise TypeError(f"request_id must be a string, got {type(request_id)}")
237

238
        # Process raw inputs into the request.
239
240
241
242
        if isinstance(prompt, EngineCoreRequest):
            request = prompt
        else:
            assert prompt_text is None
243
            request = self.input_processor.process_inputs(
244
245
246
247
248
249
250
251
252
                request_id,
                prompt,
                params,
                arrival_time,
                lora_request,
                tokenization_kwargs,
                trace_headers,
                priority,
            )
253
254
255
256
            if isinstance(prompt, str):
                prompt_text = prompt
            elif isinstance(prompt, Mapping):
                prompt_text = cast(str | None, prompt.get("prompt"))
257

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
269
270
271
272
            return

        # Fan out child requests (for n>1).
        parent_req = ParentRequest(request_id, params)
        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
    def step(self) -> list[RequestOutput | PoolingRequestOutput]:
286
287
288
289
290
        if self.should_execute_dummy_batch:
            self.should_execute_dummy_batch = False
            self.engine_core.execute_dummy_batch()
            return []

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

295
        # 2) Process EngineCoreOutputs.
296
        with record_function_or_nullcontext("llm_engine step: process_outputs"):
297
298
299
300
301
302
303
            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)
304

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

309
        # 4) Record stats
310
        with record_function_or_nullcontext("llm_engine step: record_stats"):
311
312
313
314
            if self.logger_manager is not None and outputs.scheduler_stats is not None:
                self.logger_manager.record(
                    scheduler_stats=outputs.scheduler_stats,
                    iteration_stats=iteration_stats,
315
                    mm_cache_stats=self.input_processor.stat_mm_cache(),
316
317
                )
                self.do_log_stats_with_interval()
318

319
        return processed_outputs.request_outputs
320

321
    def start_profile(self):
322
        self.engine_core.profile(True)
323

324
    def stop_profile(self):
325
        self.engine_core.profile(False)
326

327
    def reset_mm_cache(self):
328
        self.input_processor.clear_mm_cache()
329
330
        self.engine_core.reset_mm_cache()

331
332
    def reset_prefix_cache(self, reset_running_requests: bool = False) -> bool:
        return self.engine_core.reset_prefix_cache(reset_running_requests)
333

334
335
336
    def sleep(self, level: int = 1):
        self.engine_core.sleep(level)

337
338
339
        if self.logger_manager is not None:
            self.logger_manager.record_sleep_state(1, level)

340
    def wake_up(self, tags: list[str] | None = None):
341
        self.engine_core.wake_up(tags)
342

343
344
345
        if self.logger_manager is not None:
            self.logger_manager.record_sleep_state(0, 0)

346
347
348
    def is_sleeping(self) -> bool:
        return self.engine_core.is_sleeping()

349
350
351
352
    def get_metrics(self) -> list[Metric]:
        assert self.log_stats, "Stat logging disabled"
        return get_metrics_snapshot()

353
    @property
354
    def tokenizer(self) -> TokenizerLike | None:
355
        return self.input_processor.tokenizer
356
357

    @tokenizer.setter
358
    def tokenizer(self, tokenizer: TokenizerLike | None) -> None:
359
        self.input_processor.tokenizer = tokenizer
360

361
    def get_tokenizer(self) -> TokenizerLike:
362
        if self.tokenizer is None:
363
            raise ValueError(
364
                "Unable to get tokenizer because `skip_tokenizer_init=True`"
365
            )
366

367
        return self.tokenizer
368

369
370
371
372
373
374
375
376
377
378
379
380
381
382
    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

383
384
385
386
387
388
389
390
    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)

391
    def list_loras(self) -> set[int]:
392
393
394
395
396
397
        """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)
398

399
400
    def collective_rpc(
        self,
401
402
        method: str | Callable[[WorkerBase], _R],
        timeout: float | None = None,
403
        args: tuple = (),
404
        kwargs: dict[str, Any] | None = None,
405
    ) -> list[_R]:
406
407
        return self.engine_core.collective_rpc(method, timeout, args, kwargs)

408
    def apply_model(self, func: Callable[[nn.Module], _R]) -> list[_R]:
409
        return self.collective_rpc("apply_model", args=(func,))
410

411
    def __del__(self):
412
413
        dp_group = getattr(self, "dp_group", None)
        if dp_group is not None and not self.external_launcher_dp:
414
            stateless_destroy_torch_distributed_process_group(dp_group)