async_llm_engine.py 32.7 KB
Newer Older
1
2
import asyncio
import time
Antoni Baum's avatar
Antoni Baum committed
3
from functools import partial
4
5
from typing import (AsyncIterator, Callable, Dict, Iterable, List, Optional,
                    Set, Tuple, Type, Union)
6

7
8
from transformers import PreTrainedTokenizer

9
import vllm.envs as envs
10
from vllm.config import DecodingConfig, ModelConfig
11
from vllm.core.scheduler import SchedulerOutputs
Woosuk Kwon's avatar
Woosuk Kwon committed
12
13
from vllm.engine.arg_utils import AsyncEngineArgs
from vllm.engine.llm_engine import LLMEngine
14
from vllm.executor.ray_utils import initialize_ray_cluster, ray
Woosuk Kwon's avatar
Woosuk Kwon committed
15
from vllm.logger import init_logger
16
from vllm.lora.request import LoRARequest
17
18
from vllm.outputs import EmbeddingRequestOutput, RequestOutput
from vllm.pooling_params import PoolingParams
Woosuk Kwon's avatar
Woosuk Kwon committed
19
from vllm.sampling_params import SamplingParams
20
from vllm.sequence import ExecuteModelRequest, MultiModalData, SamplerOutput
yhu422's avatar
yhu422 committed
21
from vllm.usage.usage_lib import UsageContext
22
23

logger = init_logger(__name__)
24
ENGINE_ITERATION_TIMEOUT_S = envs.VLLM_ENGINE_ITERATION_TIMEOUT_S
25

Antoni Baum's avatar
Antoni Baum committed
26

27
28
29
30
class AsyncEngineDeadError(RuntimeError):
    pass


31
32
33
def _raise_exception_on_finish(
        task: asyncio.Task, error_callback: Callable[[Exception],
                                                     None]) -> None:
34
35
    msg = ("Task finished unexpectedly. This should never happen! "
           "Please open an issue on Github.")
36
37

    exception = None
38
    try:
39
40
        task.result()
        # NOTE: This will be thrown if task exits normally (which it should not)
41
        raise AsyncEngineDeadError(msg)
42
43
44
45
46
47
    except Exception as e:
        exception = e
        logger.error("Engine background task failed", exc_info=e)
        error_callback(exception)
        raise AsyncEngineDeadError(
            msg + " See stack trace above for the actual cause.") from e
48
49


Antoni Baum's avatar
Antoni Baum committed
50
class AsyncStream:
51
52
    """A stream of RequestOutputs or EmbeddingRequestOutputs for a request
    that can be iterated over asynchronously."""
Antoni Baum's avatar
Antoni Baum committed
53
54
55

    def __init__(self, request_id: str) -> None:
        self.request_id = request_id
56
        self._queue: asyncio.Queue = asyncio.Queue()
Antoni Baum's avatar
Antoni Baum committed
57
58
        self._finished = False

59
60
    def put(self, item: Union[RequestOutput, EmbeddingRequestOutput,
                              Exception]) -> None:
Antoni Baum's avatar
Antoni Baum committed
61
62
63
64
65
        if self._finished:
            return
        self._queue.put_nowait(item)

    def finish(self) -> None:
66
        self._queue.put_nowait(StopAsyncIteration())
Antoni Baum's avatar
Antoni Baum committed
67
68
69
70
71
72
73
74
75
        self._finished = True

    @property
    def finished(self) -> bool:
        return self._finished

    def __aiter__(self):
        return self

76
    async def __anext__(self) -> Union[RequestOutput, EmbeddingRequestOutput]:
Antoni Baum's avatar
Antoni Baum committed
77
        result = await self._queue.get()
78
        if isinstance(result, Exception):
79
            raise result
Antoni Baum's avatar
Antoni Baum committed
80
81
82
        return result


83
84
85
86
87
88
89
90
class RequestTracker:
    """Synchronous abstraction for tracking requests."""

    def __init__(self) -> None:
        self._request_streams: Dict[str, AsyncStream] = {}
        self._finished_requests: asyncio.Queue[str] = asyncio.Queue()
        self._new_requests: asyncio.Queue[Tuple[AsyncStream,
                                                dict]] = asyncio.Queue()
91
        self.new_requests_event = asyncio.Event()
92
93
94
95

    def __contains__(self, item):
        return item in self._request_streams

96
97
    def __len__(self) -> int:
        return len(self._request_streams)
98
99
100
101
102
103
104
105

    def propagate_exception(self,
                            exc: Exception,
                            request_id: Optional[str] = None) -> None:
        """Propagate an exception to request streams
        (all if request_id is None)."""
        if request_id is not None:
            self._request_streams[request_id].put(exc)
106
            self.abort_request(request_id)
107
        else:
108
            for rid, stream in self._request_streams.items():
109
                stream.put(exc)
110
                self.abort_request(rid)
111
112

    def process_request_output(self,
113
114
                               request_output: Union[RequestOutput,
                                                     EmbeddingRequestOutput],
115
116
117
118
119
120
121
122
                               *,
                               verbose: bool = False) -> None:
        """Process a request output from the engine."""
        request_id = request_output.request_id

        self._request_streams[request_id].put(request_output)
        if request_output.finished:
            if verbose:
123
                logger.info("Finished request %s.", request_id)
124
125
            self.abort_request(request_id)

126
127
128
129
130
131
132
133
    def process_exception(self,
                          request_id: str,
                          exception: Exception,
                          *,
                          verbose: bool = False) -> None:
        """Propagate an exception from the engine."""
        self._request_streams[request_id].put(exception)
        if verbose:
134
            logger.info("Finished request %s.", request_id)
135
136
        self.abort_request(request_id)

137
138
139
140
141
142
143
144
145
146
147
148
    def add_request(self, request_id: str,
                    **engine_add_request_kwargs) -> AsyncStream:
        """Add a request to be sent to the engine on the next background
        loop iteration."""
        if request_id in self._request_streams:
            raise KeyError(f"Request {request_id} already exists.")

        stream = AsyncStream(request_id)
        self._new_requests.put_nowait((stream, {
            "request_id": request_id,
            **engine_add_request_kwargs
        }))
149
150
151

        self.new_requests_event.set()

152
153
154
155
156
        return stream

    def abort_request(self, request_id: str, *, verbose: bool = False) -> None:
        """Abort a request during next background loop iteration."""
        if verbose:
157
            logger.info("Aborted request %s.", request_id)
158
159
160
161
162
163
164
165
166
167

        self._finished_requests.put_nowait(request_id)

        if request_id not in self._request_streams or self._request_streams[
                request_id].finished:
            # The request has already finished or been aborted.
            return

        self._request_streams[request_id].finish()

168
    def get_new_and_finished_requests(self) -> Tuple[List[Dict], Set[str]]:
169
170
        """Get the new requests and finished requests to be
        sent to the engine."""
171
        new_requests: List[Dict] = []
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
        finished_requests: Set[str] = set()

        while not self._finished_requests.empty():
            request_id = self._finished_requests.get_nowait()
            finished_requests.add(request_id)
            self._request_streams.pop(request_id, None)

        while not self._new_requests.empty():
            stream, new_request = self._new_requests.get_nowait()
            if stream.request_id in finished_requests:
                # The request has already been aborted.
                stream.finish()
                continue
            self._request_streams[stream.request_id] = stream
            new_requests.append(new_request)

        return new_requests, finished_requests
Antoni Baum's avatar
Antoni Baum committed
189

190
    async def wait_for_new_requests(self):
191
192
193
194
195
196
        if not self.has_new_requests():
            await self.new_requests_event.wait()
        self.new_requests_event.clear()

    def has_new_requests(self):
        return not self._new_requests.empty()
197

Antoni Baum's avatar
Antoni Baum committed
198
199
200
201

class _AsyncLLMEngine(LLMEngine):
    """Extension of LLMEngine to add async methods."""

202
203
    async def step_async(
            self) -> List[Union[RequestOutput, EmbeddingRequestOutput]]:
Antoni Baum's avatar
Antoni Baum committed
204
205
206
207
208
209
210
211
212
        """Performs one decoding iteration and returns newly generated results.
        The workers are ran asynchronously if possible.

        This function performs one decoding iteration of the engine. It first
        schedules the sequences to be executed in the next iteration and the
        token blocks to be swapped in/out/copy. Then, it executes the model
        and updates the scheduler with the model outputs. Finally, it decodes
        the sequences and returns the newly generated results.
        """
213
        seq_group_metadata_list, scheduler_outputs = self.scheduler.schedule()
Antoni Baum's avatar
Antoni Baum committed
214

215
216
        if not scheduler_outputs.is_empty():
            # Execute the model.
217
218
219
220
221
222
223
224
            execute_model_req = ExecuteModelRequest(
                seq_group_metadata_list=seq_group_metadata_list,
                blocks_to_swap_in=scheduler_outputs.blocks_to_swap_in,
                blocks_to_swap_out=scheduler_outputs.blocks_to_swap_out,
                blocks_to_copy=scheduler_outputs.blocks_to_copy,
                num_lookahead_slots=scheduler_outputs.num_lookahead_slots,
                running_queue_size=scheduler_outputs.running_queue_size,
            )
225
            output = await self.model_executor.execute_model_async(
226
                execute_model_req)
227
228
        else:
            output = []
Antoni Baum's avatar
Antoni Baum committed
229

230
        request_outputs = self._process_model_outputs(
231
            output, scheduler_outputs.scheduled_seq_groups,
232
            scheduler_outputs.ignored_seq_groups, seq_group_metadata_list)
Antoni Baum's avatar
Antoni Baum committed
233

234
        # Log stats.
235
        self.do_log_stats(scheduler_outputs, output)
236
237
238

        return request_outputs

239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
    async def encode_request_async(
        self,
        request_id: str,  # pylint: disable=unused-argument
        prompt: Optional[str],
        prompt_token_ids: Optional[List[int]] = None,
        lora_request: Optional[LoRARequest] = None,
    ):
        if prompt_token_ids is None:
            assert prompt is not None
            prompt_token_ids = await self.tokenizer.encode_async(
                request_id=request_id,
                prompt=prompt,
                lora_request=lora_request)
        return prompt_token_ids

    async def add_request_async(
        self,
        request_id: str,
        prompt: Optional[str],
258
        params: Union[SamplingParams, PoolingParams],
259
260
261
        prompt_token_ids: Optional[List[int]] = None,
        arrival_time: Optional[float] = None,
        lora_request: Optional[LoRARequest] = None,
262
        multi_modal_data: Optional[MultiModalData] = None,
263
264
265
266
267
268
269
270
271
272
273
274
    ) -> None:
        if lora_request is not None and not self.lora_config:
            raise ValueError(f"Got lora_request {lora_request} but LoRA is "
                             "not enabled!")
        if arrival_time is None:
            arrival_time = time.time()
        prompt_token_ids = await self.encode_request_async(
            request_id=request_id,
            prompt=prompt,
            prompt_token_ids=prompt_token_ids,
            lora_request=lora_request)

275
276
        return self.add_request(request_id,
                                prompt=prompt,
277
                                params=params,
278
279
280
281
                                prompt_token_ids=prompt_token_ids,
                                arrival_time=arrival_time,
                                lora_request=lora_request,
                                multi_modal_data=multi_modal_data)
282

283
284
    async def check_health_async(self) -> None:
        self.model_executor.check_health()
285

286

287
288
class AsyncLLMEngine:
    """An asynchronous wrapper for LLMEngine.
289

290
    This class is used to wrap the LLMEngine class to make it asynchronous. It
291
    uses asyncio to create a background loop that keeps processing incoming
292
    requests. The LLMEngine is kicked by the generate method when there
293
    are requests in the waiting queue. The generate method yields the outputs
294
    from the LLMEngine to the caller.
295

296
    NOTE: For the comprehensive list of arguments, see `LLMEngine`.
297
298
299
300
301

    Args:
        worker_use_ray: Whether to use Ray for model workers. Required for
            distributed execution. Should be the same as
            `parallel_config.worker_use_ray`.
Zhuohan Li's avatar
Zhuohan Li committed
302
        engine_use_ray: Whether to make LLMEngine a Ray actor. If so, the
303
304
            async frontend will be executed in a separate process as the
            model workers.
305
        log_requests: Whether to log the requests.
zspo's avatar
zspo committed
306
307
        max_log_len: Maximum number of prompt characters or prompt ID numbers
            being printed in log.
308
309
        start_engine_loop: If True, the background task to run the engine
            will be automatically started in the generate call.
310
311
        *args: Arguments for LLMEngine.
        *kwargs: Arguments for LLMEngine.
312
    """
313

Antoni Baum's avatar
Antoni Baum committed
314
315
    _engine_class: Type[_AsyncLLMEngine] = _AsyncLLMEngine

316
317
318
319
320
    def __init__(self,
                 worker_use_ray: bool,
                 engine_use_ray: bool,
                 *args,
                 log_requests: bool = True,
321
                 max_log_len: Optional[int] = None,
322
                 start_engine_loop: bool = True,
323
                 **kwargs) -> None:
324
        self.worker_use_ray = worker_use_ray
Zhuohan Li's avatar
Zhuohan Li committed
325
        self.engine_use_ray = engine_use_ray
326
        self.log_requests = log_requests
327
        self.max_log_len = max_log_len
Antoni Baum's avatar
Antoni Baum committed
328
329
        self.engine = self._init_engine(*args, **kwargs)

330
        self.background_loop: Optional[asyncio.Future] = None
331
332
333
        # We need to keep a reference to unshielded
        # task as well to prevent it from being garbage
        # collected
334
        self._background_loop_unshielded: Optional[asyncio.Task] = None
335
        self.start_engine_loop = start_engine_loop
336
        self._errored_with: Optional[BaseException] = None
Antoni Baum's avatar
Antoni Baum committed
337

338
339
340
        # Lazy initialized fields
        self._request_tracker: RequestTracker

341
    @classmethod
yhu422's avatar
yhu422 committed
342
343
344
345
346
347
    def from_engine_args(
        cls,
        engine_args: AsyncEngineArgs,
        start_engine_loop: bool = True,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
    ) -> "AsyncLLMEngine":
348
349
        """Creates an async LLM engine from the engine arguments."""
        # Create the engine configs.
350
        engine_config = engine_args.create_engine_config()
351
352
        distributed_executor_backend = (
            engine_config.parallel_config.distributed_executor_backend)
353

354
        if engine_config.device_config.device_type == "neuron":
355
356
            from vllm.executor.neuron_executor import NeuronExecutorAsync
            executor_class = NeuronExecutorAsync
357
        elif engine_config.device_config.device_type == "cpu":
358
359
            assert distributed_executor_backend is None, (
                "Distributed execution is not supported with the CPU backend.")
360
361
            from vllm.executor.cpu_executor import CPUExecutorAsync
            executor_class = CPUExecutorAsync
362
        elif distributed_executor_backend == "ray":
363
            initialize_ray_cluster(engine_config.parallel_config)
364
365
            from vllm.executor.ray_gpu_executor import RayGPUExecutorAsync
            executor_class = RayGPUExecutorAsync
366
367
368
369
        elif distributed_executor_backend == "mp":
            from vllm.executor.multiproc_gpu_executor import (
                MultiprocessingGPUExecutorAsync)
            executor_class = MultiprocessingGPUExecutorAsync
370
371
372
373
        else:
            from vllm.executor.gpu_executor import GPUExecutorAsync
            executor_class = GPUExecutorAsync
        # Create the async LLM engine.
yhu422's avatar
yhu422 committed
374
        engine = cls(
375
            distributed_executor_backend == "ray",
yhu422's avatar
yhu422 committed
376
            engine_args.engine_use_ray,
377
378
            **engine_config.to_dict(),
            executor_class=executor_class,
yhu422's avatar
yhu422 committed
379
380
381
382
383
384
            log_requests=not engine_args.disable_log_requests,
            log_stats=not engine_args.disable_log_stats,
            max_log_len=engine_args.max_log_len,
            start_engine_loop=start_engine_loop,
            usage_context=usage_context,
        )
385
386
        return engine

387
388
    @property
    def is_running(self) -> bool:
389
        return (self.background_loop is not None
390
                and self._background_loop_unshielded is not None
391
392
393
394
                and not self._background_loop_unshielded.done())

    @property
    def is_stopped(self) -> bool:
395
396
        return self.errored or (self.background_loop is not None and
                                self._background_loop_unshielded is not None
397
398
399
400
401
402
403
404
405
406
407
408
                                and self._background_loop_unshielded.done())

    @property
    def errored(self) -> bool:
        return self._errored_with is not None

    def set_errored(self, exc: Exception) -> None:
        self._errored_with = exc

    def _error_callback(self, exc: Exception) -> None:
        self.set_errored(exc)
        self._request_tracker.propagate_exception(exc)
409

410
411
    async def get_tokenizer(self) -> "PreTrainedTokenizer":
        if self.engine_use_ray:
412
            return await self.engine.get_tokenizer.remote()  # type: ignore
413
414
        else:
            return self.engine.get_tokenizer()
415

416
    def start_background_loop(self) -> None:
Antoni Baum's avatar
Antoni Baum committed
417
        """Start the background loop."""
418
419
420
        if self.errored:
            raise AsyncEngineDeadError(
                "Background loop has errored already.") from self._errored_with
421
        if self.is_running:
Antoni Baum's avatar
Antoni Baum committed
422
            raise RuntimeError("Background loop is already running.")
423
424
        # Initialize the RequestTracker here so it uses the right event loop.
        self._request_tracker = RequestTracker()
425
426
427
428

        self._background_loop_unshielded = asyncio.get_event_loop(
        ).create_task(self.run_engine_loop())
        self._background_loop_unshielded.add_done_callback(
429
            partial(_raise_exception_on_finish,
430
                    error_callback=self._error_callback))
431
        self.background_loop = asyncio.shield(self._background_loop_unshielded)
Antoni Baum's avatar
Antoni Baum committed
432
433
434

    def _init_engine(self, *args,
                     **kwargs) -> Union[_AsyncLLMEngine, "ray.ObjectRef"]:
Zhuohan Li's avatar
Zhuohan Li committed
435
        if not self.engine_use_ray:
Antoni Baum's avatar
Antoni Baum committed
436
            engine_class = self._engine_class
437
        elif self.worker_use_ray:
Antoni Baum's avatar
Antoni Baum committed
438
            engine_class = ray.remote(num_cpus=0)(self._engine_class).remote
439
        else:
Woosuk Kwon's avatar
Woosuk Kwon committed
440
441
            # FIXME(woosuk): This is a bit hacky. Be careful when changing the
            # order of the arguments.
442
443
            cache_config = kwargs["cache_config"]
            parallel_config = kwargs["parallel_config"]
Woosuk Kwon's avatar
Woosuk Kwon committed
444
445
446
447
448
449
            if parallel_config.tensor_parallel_size == 1:
                num_gpus = cache_config.gpu_memory_utilization
            else:
                num_gpus = 1
            engine_class = ray.remote(num_gpus=num_gpus)(
                self._engine_class).remote
Antoni Baum's avatar
Antoni Baum committed
450
451
        return engine_class(*args, **kwargs)

452
453
454
455
    async def engine_step(self) -> bool:
        """Kick the engine to process the waiting requests.

        Returns True if there are in-progress requests."""
456
457

        new_requests, finished_requests = (
458
            self._request_tracker.get_new_and_finished_requests())
459
460
461
462

        for new_request in new_requests:
            # Add the request into the vLLM engine's waiting queue.
            # TODO: Maybe add add_request_batch to reduce Ray overhead
463
464
            try:
                if self.engine_use_ray:
465
466
                    await self.engine.add_request.remote(  # type: ignore
                        **new_request)
467
468
469
470
471
472
473
474
475
                else:
                    await self.engine.add_request_async(**new_request)
            except ValueError as e:
                # TODO: use a vLLM specific error for failed validation
                self._request_tracker.process_exception(
                    new_request["request_id"],
                    e,
                    verbose=self.log_requests,
                )
476
477
478
479

        if finished_requests:
            await self._engine_abort(finished_requests)

Zhuohan Li's avatar
Zhuohan Li committed
480
        if self.engine_use_ray:
481
            request_outputs = await self.engine.step.remote()  # type: ignore
482
        else:
Antoni Baum's avatar
Antoni Baum committed
483
            request_outputs = await self.engine.step_async()
484

Antoni Baum's avatar
Antoni Baum committed
485
        # Put the outputs into the corresponding streams.
486
        for request_output in request_outputs:
487
            self._request_tracker.process_request_output(
488
                request_output, verbose=self.log_requests)
Antoni Baum's avatar
Antoni Baum committed
489

490
491
        return len(request_outputs) > 0

Antoni Baum's avatar
Antoni Baum committed
492
493
    async def _engine_abort(self, request_ids: Iterable[str]):
        if self.engine_use_ray:
494
            await self.engine.abort_request.remote(request_ids)  # type: ignore
Antoni Baum's avatar
Antoni Baum committed
495
496
497
498
        else:
            self.engine.abort_request(request_ids)

    async def run_engine_loop(self):
499
        has_requests_in_progress = False
Antoni Baum's avatar
Antoni Baum committed
500
        while True:
501
            if not has_requests_in_progress:
502
                logger.debug("Waiting for new requests...")
503
                await self._request_tracker.wait_for_new_requests()
504
505
506
507
508
509
510
511
512
513
514
515
                logger.debug("Got new requests!")

            # Abort if iteration takes too long due to unrecoverable errors
            # (eg. NCCL timeouts).
            try:
                has_requests_in_progress = await asyncio.wait_for(
                    self.engine_step(), ENGINE_ITERATION_TIMEOUT_S)
            except asyncio.TimeoutError as exc:
                logger.error(
                    "Engine iteration timed out. This should never happen!")
                self.set_errored(exc)
                raise
Antoni Baum's avatar
Antoni Baum committed
516
517
518
519
520
521
            await asyncio.sleep(0)

    async def add_request(
        self,
        request_id: str,
        prompt: Optional[str],
522
        params: Union[SamplingParams, PoolingParams],
Antoni Baum's avatar
Antoni Baum committed
523
524
        prompt_token_ids: Optional[List[int]] = None,
        arrival_time: Optional[float] = None,
525
        lora_request: Optional[LoRARequest] = None,
526
        multi_modal_data: Optional[MultiModalData] = None,
Antoni Baum's avatar
Antoni Baum committed
527
528
    ) -> AsyncStream:
        if self.log_requests:
529
530
531
532
533
534
535
536
            shortened_prompt = prompt
            shortened_token_ids = prompt_token_ids
            if self.max_log_len is not None:
                if shortened_prompt is not None:
                    shortened_prompt = shortened_prompt[:self.max_log_len]
                if shortened_token_ids is not None:
                    shortened_token_ids = shortened_token_ids[:self.
                                                              max_log_len]
537
538
            logger.info(
                "Received request %s: prompt: %r, "
539
540
541
                "params: %s, prompt_token_ids: %s, "
                "lora_request: %s.", request_id, shortened_prompt, params,
                shortened_token_ids, lora_request)
Antoni Baum's avatar
Antoni Baum committed
542

543
        if not self.is_running:
544
545
546
547
548
549
550
551
            if self.start_engine_loop:
                self.start_background_loop()
            else:
                raise AsyncEngineDeadError(
                    "Background loop is not running. If it was running, "
                    "inspect the output to find the stacktrace of the "
                    "error that caused the background loop to stop "
                    "(AsyncEngineDeadError).")
Antoni Baum's avatar
Antoni Baum committed
552

553
554
        if arrival_time is None:
            arrival_time = time.time()
555
556

        if self.engine_use_ray:
557
558
559
560
561
562
            prompt_token_ids = await (
                self.engine.encode_request_async.remote(  # type: ignore
                    request_id=request_id,
                    prompt=prompt,
                    prompt_token_ids=prompt_token_ids,
                    lora_request=lora_request))
563
564
565
566
567
568
        else:
            prompt_token_ids = await self.engine.encode_request_async(
                request_id=request_id,
                prompt=prompt,
                prompt_token_ids=prompt_token_ids,
                lora_request=lora_request)
569

570
        stream = self._request_tracker.add_request(
571
572
            request_id,
            prompt=prompt,
573
            params=params,
574
            prompt_token_ids=prompt_token_ids,
575
            arrival_time=arrival_time,
576
577
578
            lora_request=lora_request,
            multi_modal_data=multi_modal_data,
        )
Antoni Baum's avatar
Antoni Baum committed
579
580

        return stream
581

582
    async def generate(
583
584
585
586
        self,
        prompt: Optional[str],
        sampling_params: SamplingParams,
        request_id: str,
587
        prompt_token_ids: Optional[List[int]] = None,
588
        lora_request: Optional[LoRARequest] = None,
589
        multi_modal_data: Optional[MultiModalData] = None
590
    ) -> AsyncIterator[RequestOutput]:
591
592
593
        """Generate outputs for a request.

        Generate outputs for a request. This method is a coroutine. It adds the
594
595
        request into the waiting queue of the LLMEngine and streams the outputs
        from the LLMEngine to the caller.
596
597
598
599
600
601
602
603

        Args:
            prompt: The prompt string. Can be None if prompt_token_ids is
                provided.
            sampling_params: The sampling parameters of the request.
            request_id: The unique id of the request.
            prompt_token_ids: The token IDs of the prompt. If None, we
                use the tokenizer to convert the prompts to token IDs.
604
            lora_request: LoRA request to use for generation, if any.
605
            multi_modal_data: Multi modal data per request.
606
607

        Yields:
608
609
            The output `RequestOutput` objects from the LLMEngine
            for the request.
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652

        Details:
            - If the engine is not running, start the background loop,
              which iteratively invokes
              :meth:`~vllm.engine.async_llm_engine.AsyncLLMEngine.engine_step`
              to process the waiting requests.
            - Add the request to the engine's `RequestTracker`.
              On the next background loop, this request will be sent to
              the underlying engine.
              Also, a corresponding `AsyncStream` will be created.
            - Wait for the request outputs from `AsyncStream` and yield them.

        Example:
            >>> # Please refer to entrypoints/api_server.py for
            >>> # the complete example.
            >>>
            >>> # initialize the engine and the example input
            >>> engine = AsyncLLMEngine.from_engine_args(engine_args)
            >>> example_input = {
            >>>     "prompt": "What is LLM?",
            >>>     "stream": False, # assume the non-streaming case
            >>>     "temperature": 0.0,
            >>>     "request_id": 0,
            >>> }
            >>>
            >>> # start the generation
            >>> results_generator = engine.generate(
            >>>    example_input["prompt"],
            >>>    SamplingParams(temperature=example_input["temperature"]),
            >>>    example_input["request_id"])
            >>>
            >>> # get the results
            >>> final_output = None
            >>> async for request_output in results_generator:
            >>>     if await request.is_disconnected():
            >>>         # Abort the request if the client disconnects.
            >>>         await engine.abort(request_id)
            >>>         # Return or raise an error
            >>>         ...
            >>>     final_output = request_output
            >>>
            >>> # Process and return the final output
            >>> ...
653
        """
654
        async for output in self.process_request(
655
656
657
                request_id,
                prompt,
                sampling_params,
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
                prompt_token_ids,
                lora_request,
                multi_modal_data,
        ):
            yield output

    async def encode(
        self,
        prompt: Optional[str],
        pooling_params: PoolingParams,
        request_id: str,
        prompt_token_ids: Optional[List[int]] = None,
        lora_request: Optional[LoRARequest] = None,
        multi_modal_data: Optional[MultiModalData] = None
    ) -> AsyncIterator[EmbeddingRequestOutput]:
        """Generate outputs for a request from an embedding model.

        Generate outputs for a request. This method is a coroutine. It adds the
        request into the waiting queue of the LLMEngine and streams the outputs
        from the LLMEngine to the caller.

        Args:
            prompt: The prompt string. Can be None if prompt_token_ids is
                provided.
            pooling_params: The pooling parameters of the request.
            request_id: The unique id of the request.
            prompt_token_ids: The token IDs of the prompt. If None, we
                use the tokenizer to convert the prompts to token IDs.
            lora_request: LoRA request to use for generation, if any.
            multi_modal_data: Multi modal data per request.

        Yields:
            The output `EmbeddingRequestOutput` objects from the LLMEngine 
            for the request.

        Details:
            - If the engine is not running, start the background loop,
              which iteratively invokes
              :meth:`~vllm.engine.async_llm_engine.AsyncLLMEngine.engine_step`
              to process the waiting requests.
            - Add the request to the engine's `RequestTracker`.
              On the next background loop, this request will be sent to
              the underlying engine.
              Also, a corresponding `AsyncStream` will be created.
            - Wait for the request outputs from `AsyncStream` and yield them.

        Example:
            >>> # Please refer to entrypoints/api_server.py for
            >>> # the complete example.
            >>>
            >>> # initialize the engine and the example input
            >>> engine = AsyncLLMEngine.from_engine_args(engine_args)
            >>> example_input = {
            >>>     "input": "What is LLM?",
            >>>     "request_id": 0,
            >>> }
            >>>
            >>> # start the generation
            >>> results_generator = engine.encode(
            >>>    example_input["input"],
            >>>    PoolingParams(),
            >>>    example_input["request_id"])
            >>>
            >>> # get the results
            >>> final_output = None
            >>> async for request_output in results_generator:
            >>>     if await request.is_disconnected():
            >>>         # Abort the request if the client disconnects.
            >>>         await engine.abort(request_id)
            >>>         # Return or raise an error
            >>>         ...
            >>>     final_output = request_output
            >>>
            >>> # Process and return the final output
            >>> ...
        """
        async for output in self.process_request(
                request_id,
                prompt,
                pooling_params,
                prompt_token_ids,
                lora_request,
                multi_modal_data,
        ):
            yield output

    async def process_request(
        self,
        request_id: str,
        prompt: Optional[str],
        params: Union[SamplingParams, PoolingParams],
        prompt_token_ids: Optional[List[int]] = None,
        lora_request: Optional[LoRARequest] = None,
        multi_modal_data: Optional[MultiModalData] = None,
    ) -> AsyncIterator[Union[RequestOutput, EmbeddingRequestOutput]]:
        """Common logic to process requests with SamplingParams or
        PoolingParams."""
        arrival_time = time.time()

        stream = await self.add_request(
            request_id,
            prompt,
            params,
            prompt_token_ids=prompt_token_ids,
            arrival_time=arrival_time,
            lora_request=lora_request,
            multi_modal_data=multi_modal_data,
        )
766

767
        try:
Antoni Baum's avatar
Antoni Baum committed
768
769
            async for request_output in stream:
                yield request_output
770
        except (Exception, asyncio.CancelledError) as e:
Antoni Baum's avatar
Antoni Baum committed
771
772
            self._abort(request_id)
            raise e
773

Antoni Baum's avatar
Antoni Baum committed
774
775
    async def abort(self, request_id: str) -> None:
        """Abort a request.
776

Antoni Baum's avatar
Antoni Baum committed
777
778
        Abort a submitted request. If the request is finished or not found,
        this method will be a no-op.
779

Antoni Baum's avatar
Antoni Baum committed
780
781
782
        Args:
            request_id: The unique id of the request.
        """
783
784
785
786
787
788
789
        if not self.is_running:
            raise AsyncEngineDeadError(
                "Background loop is not running. If it was running, "
                "inspect the output to find the stacktrace of the "
                "error that caused the background loop to stop "
                "(AsyncEngineDeadError).")

Antoni Baum's avatar
Antoni Baum committed
790
        return self._abort(request_id)
791

Antoni Baum's avatar
Antoni Baum committed
792
    def _abort(self, request_id: str) -> None:
793
794
795
796
797
798
799
800
        """Abort a request.

        Abort a submitted request. If the request is finished or not found,
        this method will be a no-op.

        Args:
            request_id: The unique id of the request.
        """
801
802
        self._request_tracker.abort_request(request_id,
                                            verbose=self.log_requests)
803

804
805
806
    async def get_model_config(self) -> ModelConfig:
        """Get the model configuration of the vLLM engine."""
        if self.engine_use_ray:
807
            return await self.engine.get_model_config.remote()  # type: ignore
808
809
810
        else:
            return self.engine.get_model_config()

811
812
813
814
815
816
817
818
    async def get_decoding_config(self) -> DecodingConfig:
        """Get the decoding configuration of the vLLM engine."""
        if self.engine_use_ray:
            return await self.engine.get_decoding_config.remote(  # type: ignore
            )
        else:
            return self.engine.get_decoding_config()

819
820
821
822
    async def do_log_stats(
            self,
            scheduler_outputs: Optional[SchedulerOutputs] = None,
            model_output: Optional[List[SamplerOutput]] = None) -> None:
823
        if self.engine_use_ray:
824
825
            await self.engine.do_log_stats.remote(  # type: ignore
                scheduler_outputs, model_output)
826
827
        else:
            self.engine.do_log_stats()
828

829
    async def check_health(self) -> None:
830
831
832
833
834
835
836
837
        """Raises an error if engine is unhealthy."""
        t = time.perf_counter()
        logger.debug("Starting health check...")
        if self.is_stopped:
            raise AsyncEngineDeadError("Background loop is stopped.")

        if self.engine_use_ray:
            try:
838
                await self.engine.check_health.remote()  # type: ignore
839
840
841
842
            except ray.exceptions.RayActorError as e:
                raise RuntimeError("Engine is dead.") from e
        else:
            await self.engine.check_health_async()
843
        logger.debug("Health check took %fs", time.perf_counter() - t)