llm_engine.py 16.2 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
        use_cached_outputs: bool = False,
60
        multiprocess_mode: bool = False,
61
    ) -> None:
62
        self.vllm_config = vllm_config
63
        self.model_config = vllm_config.model_config
64
65
66
67
68
        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)
69

70
71
        self.log_stats = log_stats

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

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

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

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

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

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

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

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

127
128
129
130
131
        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

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

135
136
137
138
139
    @classmethod
    def from_vllm_config(
        cls,
        vllm_config: VllmConfig,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
140
        stat_loggers: list[StatLoggerFactory] | None = None,
141
142
        disable_log_stats: bool = False,
    ) -> "LLMEngine":
143
144
145
146
147
148
149
150
        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,
        )
151

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

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

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

        # Create the LLMEngine.
171
172
173
174
175
176
177
178
        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,
        )
179
180

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

    def has_unfinished_requests(self) -> bool:
184
        has_unfinished = self.output_processor.has_unfinished_requests()
185
        if self.dp_group is None:
186
            return has_unfinished or self.engine_core.dp_engines_running()
187
188
189
190
        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(
191
192
            self.dp_group, has_unfinished
        )
193
194
195
        if not has_unfinished and aggregated_has_unfinished:
            self.should_execute_dummy_batch = True
        return aggregated_has_unfinished
196

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

        return self._supported_tasks
203

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

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

210
211
212
    def add_request(
        self,
        request_id: str,
213
        prompt: EngineCoreRequest | PromptType | EngineInput,
214
215
216
217
218
        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,
219
        priority: int = 0,
220
        prompt_text: str | None = None,
221
    ) -> str:
222
223
        # Validate the request_id type.
        if not isinstance(request_id, str):
224
            raise TypeError(f"request_id must be a string, got {type(request_id)}")
225

226
        # Process raw inputs into the request.
227
        if isinstance(prompt, EngineCoreRequest):
228
229
230
231
232
233
            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()."
            )

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

255
256
        self.input_processor.assign_request_id(request)

257
258
        req_id = request.request_id

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

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

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

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

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

286
287
        return req_id

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

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

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

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

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

326
        return processed_outputs.request_outputs
327

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

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

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

338
339
340
341
342
343
    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
        )
344

345
346
347
348
349
350
351
352
    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()

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

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

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

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

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

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

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

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

379
380
381
382
383
384
385
386
387
388
389
390
391
392
    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

393
394
395
396
397
398
399
400
    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)

401
    def list_loras(self) -> set[int]:
402
403
404
405
406
407
        """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)
408

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

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

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