Skip to content

vllm.engine.protocol

logger module-attribute 

logger = init_logger(__name__)

EngineClient

Bases: ABC

Protocol class for Clients to Engine

Source code in vllm/engine/protocol.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
class EngineClient(ABC):
    """Protocol class for Clients to Engine"""

    @property
    @abstractmethod
    def is_running(self) -> bool:
        ...

    @property
    @abstractmethod
    def is_stopped(self) -> bool:
        ...

    @property
    @abstractmethod
    def errored(self) -> bool:
        ...

    @property
    @abstractmethod
    def dead_error(self) -> BaseException:
        ...

    @abstractmethod
    def generate(
        self,
        prompt: PromptType,
        sampling_params: SamplingParams,
        request_id: str,
        lora_request: Optional[LoRARequest] = None,
        trace_headers: Optional[Mapping[str, str]] = None,
        prompt_adapter_request: Optional[PromptAdapterRequest] = None,
        priority: int = 0,
    ) -> AsyncGenerator[RequestOutput, None]:
        """Generate outputs for a request."""
        ...

    async def beam_search(
        self,
        prompt: PromptType,
        request_id: str,
        params: BeamSearchParams,
        lora_request: Optional[LoRARequest] = None,
    ) -> AsyncGenerator[RequestOutput, None]:

        beam_width = params.beam_width
        max_tokens = params.max_tokens
        ignore_eos = params.ignore_eos
        temperature = params.temperature
        length_penalty = params.length_penalty
        include_stop_str_in_output = params.include_stop_str_in_output

        preprocessor = await self.get_input_preprocessor()
        tokenizer_group = preprocessor.get_tokenizer_group()
        tokenizer = await tokenizer_group.get_lora_tokenizer_async()

        if is_explicit_encoder_decoder_prompt(prompt):
            raise NotImplementedError
        else:
            processed_inputs = preprocessor._prompt_to_llm_inputs(prompt)

        if processed_inputs["type"] == "embeds":
            raise NotImplementedError

        # This is a workaround to fix multimodal beam search; this is a
        # bandaid fix for 2 small problems:
        # 1. Multi_modal_data on the processed_inputs currently resolves to
        #    `None`.
        # 2. preprocessing above expands the multimodal placeholders. However,
        #    this happens again in generation, so the double expansion causes
        #    a mismatch.
        # TODO - would be ideal to handle this more gracefully.
        prompt_token_ids = prompt.get("prompt_token_ids")
        multi_modal_data = prompt.get("multi_modal_data")

        prompt_text = processed_inputs.get("prompt")
        mm_processor_kwargs = processed_inputs.get("mm_processor_kwargs")

        tokenized_length = len(prompt_token_ids)

        sort_beams_key = create_sort_beams_key_function(
            tokenizer.eos_token_id, length_penalty)

        beam_search_params = SamplingParams(
            logprobs=2 * beam_width,
            max_tokens=1,
            temperature=temperature,
        )
        all_beams = [
            BeamSearchSequence(tokens=prompt_token_ids,
                               cum_logprob=0,
                               logprobs=[],
                               multi_modal_data=multi_modal_data,
                               mm_processor_kwargs=mm_processor_kwargs,
                               lora_request=lora_request)
        ]
        completed = []

        for _ in range(max_tokens):
            prompts_batch, lora_req_batch = zip(*[(
                TokensPrompt(prompt_token_ids=beam.tokens,
                             multi_modal_data=beam.multi_modal_data,
                             mm_processor_kwargs=beam.mm_processor_kwargs),
                beam.lora_request,
            ) for beam in all_beams])

            tasks = []

            request_id = f"beam_search-{random_uuid()}"
            for i, (individual_prompt,
                    lora_req) in enumerate(zip(prompts_batch, lora_req_batch)):
                request_id_item = f"{request_id}-{i}"
                task = asyncio.create_task(
                    collect_from_async_generator(
                        self.generate(individual_prompt,
                                      beam_search_params,
                                      request_id_item,
                                      lora_request=lora_req)))
                tasks.append(task)

            output = await asyncio.gather(*tasks)

            output = [x[0] for x in output]

            new_beams = []
            for i, current_beam in enumerate(all_beams):
                result = output[i]

                if result.outputs[0].logprobs is not None:
                    logprobs = result.outputs[0].logprobs[0]
                    for token_id, logprob_obj in logprobs.items():
                        if token_id == tokenizer.eos_token_id and \
                            not ignore_eos:
                            completed.append(
                                BeamSearchSequence(
                                    tokens=current_beam.tokens +
                                    [token_id] if include_stop_str_in_output
                                    else current_beam.tokens,
                                    logprobs=current_beam.logprobs +
                                    [logprobs],
                                    cum_logprob=current_beam.cum_logprob +
                                    logprob_obj.logprob,
                                    finish_reason="stop",
                                    stop_reason=tokenizer.eos_token_id))
                        else:
                            new_beams.append(
                                BeamSearchSequence(
                                    tokens=current_beam.tokens + [token_id],
                                    logprobs=current_beam.logprobs +
                                    [logprobs],
                                    lora_request=current_beam.lora_request,
                                    cum_logprob=current_beam.cum_logprob +
                                    logprob_obj.logprob,
                                    multi_modal_data=current_beam.
                                    multi_modal_data,
                                    mm_processor_kwargs=current_beam.
                                    mm_processor_kwargs))

            sorted_beams = sorted(new_beams, key=sort_beams_key, reverse=True)
            all_beams = sorted_beams[:beam_width]

        completed.extend(all_beams)
        sorted_completed = sorted(completed, key=sort_beams_key, reverse=True)
        best_beams = sorted_completed[:beam_width]

        for beam in best_beams:
            if (beam.tokens[-1] == tokenizer.eos_token_id and not ignore_eos):
                # Skip the eos token in the text.
                tokens = beam.tokens[tokenized_length:-1]
            else:
                tokens = beam.tokens[tokenized_length:]
            beam.text = tokenizer.decode(tokens)

        beam_search_output = RequestOutput(
            request_id=request_id,
            prompt=prompt_text,
            outputs=[
                CompletionOutput(text=beam.text,
                                 cumulative_logprob=beam.cum_logprob,
                                 token_ids=beam.tokens[tokenized_length:],
                                 index=i,
                                 logprobs=beam.logprobs,
                                 finish_reason=beam.finish_reason if
                                 beam.finish_reason is not None else "length",
                                 stop_reason=beam.stop_reason)
                for (i, beam) in enumerate(best_beams)
            ],
            finished=True,
            prompt_token_ids=prompt_token_ids,
            prompt_logprobs=None)

        yield beam_search_output

    @abstractmethod
    def encode(
        self,
        prompt: PromptType,
        pooling_params: PoolingParams,
        request_id: str,
        lora_request: Optional[LoRARequest] = None,
        trace_headers: Optional[Mapping[str, str]] = None,
        priority: int = 0,
    ) -> AsyncGenerator[PoolingRequestOutput, None]:
        """Generate outputs for a request from a pooling model."""
        ...

    @abstractmethod
    async def abort(self, request_id: str) -> None:
        """Abort a request.

        Args:
            request_id: The unique id of the request.
        """
        ...

    @abstractmethod
    async def get_vllm_config(self) -> VllmConfig:
        """Get the vllm configuration of the vLLM engine."""
        ...

    @abstractmethod
    async def get_model_config(self) -> ModelConfig:
        """Get the model configuration of the vLLM engine."""
        ...

    @abstractmethod
    async def get_decoding_config(self) -> DecodingConfig:
        """Get the decoding configuration of the vLLM engine."""
        ...

    @abstractmethod
    async def get_input_preprocessor(self) -> InputPreprocessor:
        """Get the input processor of the vLLM engine."""
        ...

    @abstractmethod
    async def get_tokenizer(
        self,
        lora_request: Optional[LoRARequest] = None,
    ) -> AnyTokenizer:
        """Get the appropriate tokenizer for the request"""
        ...

    @abstractmethod
    async def is_tracing_enabled(self) -> bool:
        ...

    @abstractmethod
    async def do_log_stats(
        self,
        scheduler_outputs: Optional[SchedulerOutputs] = None,
        model_output: Optional[list[SamplerOutput]] = None,
    ) -> None:
        ...

    @abstractmethod
    async def check_health(self) -> None:
        """Raise if unhealthy"""
        ...

    @abstractmethod
    async def start_profile(self) -> None:
        """Start profiling the engine"""
        ...

    @abstractmethod
    async def stop_profile(self) -> None:
        """Start profiling the engine"""
        ...

    @abstractmethod
    async def reset_mm_cache(self) -> None:
        """Reset the multi-modal cache"""
        ...

    @abstractmethod
    async def reset_prefix_cache(self,
                                 device: Optional[Device] = None) -> None:
        """Reset the prefix cache"""
        ...

    @abstractmethod
    async def sleep(self, level: int = 1) -> None:
        """Sleep the engine"""
        ...

    @abstractmethod
    async def wake_up(self, tags: Optional[list[str]] = None) -> None:
        """Wake up the engine"""
        ...

    @abstractmethod
    async def is_sleeping(self) -> bool:
        """Check whether the engine is sleeping"""
        ...

    @abstractmethod
    async def add_lora(self, lora_request: LoRARequest) -> None:
        """Load a new LoRA adapter into the engine for future requests."""
        ...

dead_error abstractmethod property 

dead_error: BaseException

errored abstractmethod property 

errored: bool

is_running abstractmethod property 

is_running: bool

is_stopped abstractmethod property 

is_stopped: bool

abort abstractmethod async 

abort(request_id: str) -> None

Abort a request.

Parameters:

Name Type Description Default
request_id str

The unique id of the request.

 required
Source code in vllm/engine/protocol.py 
233
234
235
236
237
238
239
240
@abstractmethod
async def abort(self, request_id: str) -> None:
    """Abort a request.

    Args:
        request_id: The unique id of the request.
    """
    ...

add_lora abstractmethod async 

add_lora(lora_request: LoRARequest) -> None

Load a new LoRA adapter into the engine for future requests.

Source code in vllm/engine/protocol.py 
323
324
325
326
@abstractmethod
async def add_lora(self, lora_request: LoRARequest) -> None:
    """Load a new LoRA adapter into the engine for future requests."""
    ...

beam_search(
    prompt: PromptType,
    request_id: str,
    params: BeamSearchParams,
    lora_request: Optional[LoRARequest] = None,
) -> AsyncGenerator[RequestOutput, None]
Source code in vllm/engine/protocol.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
async def beam_search(
    self,
    prompt: PromptType,
    request_id: str,
    params: BeamSearchParams,
    lora_request: Optional[LoRARequest] = None,
) -> AsyncGenerator[RequestOutput, None]:

    beam_width = params.beam_width
    max_tokens = params.max_tokens
    ignore_eos = params.ignore_eos
    temperature = params.temperature
    length_penalty = params.length_penalty
    include_stop_str_in_output = params.include_stop_str_in_output

    preprocessor = await self.get_input_preprocessor()
    tokenizer_group = preprocessor.get_tokenizer_group()
    tokenizer = await tokenizer_group.get_lora_tokenizer_async()

    if is_explicit_encoder_decoder_prompt(prompt):
        raise NotImplementedError
    else:
        processed_inputs = preprocessor._prompt_to_llm_inputs(prompt)

    if processed_inputs["type"] == "embeds":
        raise NotImplementedError

    # This is a workaround to fix multimodal beam search; this is a
    # bandaid fix for 2 small problems:
    # 1. Multi_modal_data on the processed_inputs currently resolves to
    #    `None`.
    # 2. preprocessing above expands the multimodal placeholders. However,
    #    this happens again in generation, so the double expansion causes
    #    a mismatch.
    # TODO - would be ideal to handle this more gracefully.
    prompt_token_ids = prompt.get("prompt_token_ids")
    multi_modal_data = prompt.get("multi_modal_data")

    prompt_text = processed_inputs.get("prompt")
    mm_processor_kwargs = processed_inputs.get("mm_processor_kwargs")

    tokenized_length = len(prompt_token_ids)

    sort_beams_key = create_sort_beams_key_function(
        tokenizer.eos_token_id, length_penalty)

    beam_search_params = SamplingParams(
        logprobs=2 * beam_width,
        max_tokens=1,
        temperature=temperature,
    )
    all_beams = [
        BeamSearchSequence(tokens=prompt_token_ids,
                           cum_logprob=0,
                           logprobs=[],
                           multi_modal_data=multi_modal_data,
                           mm_processor_kwargs=mm_processor_kwargs,
                           lora_request=lora_request)
    ]
    completed = []

    for _ in range(max_tokens):
        prompts_batch, lora_req_batch = zip(*[(
            TokensPrompt(prompt_token_ids=beam.tokens,
                         multi_modal_data=beam.multi_modal_data,
                         mm_processor_kwargs=beam.mm_processor_kwargs),
            beam.lora_request,
        ) for beam in all_beams])

        tasks = []

        request_id = f"beam_search-{random_uuid()}"
        for i, (individual_prompt,
                lora_req) in enumerate(zip(prompts_batch, lora_req_batch)):
            request_id_item = f"{request_id}-{i}"
            task = asyncio.create_task(
                collect_from_async_generator(
                    self.generate(individual_prompt,
                                  beam_search_params,
                                  request_id_item,
                                  lora_request=lora_req)))
            tasks.append(task)

        output = await asyncio.gather(*tasks)

        output = [x[0] for x in output]

        new_beams = []
        for i, current_beam in enumerate(all_beams):
            result = output[i]

            if result.outputs[0].logprobs is not None:
                logprobs = result.outputs[0].logprobs[0]
                for token_id, logprob_obj in logprobs.items():
                    if token_id == tokenizer.eos_token_id and \
                        not ignore_eos:
                        completed.append(
                            BeamSearchSequence(
                                tokens=current_beam.tokens +
                                [token_id] if include_stop_str_in_output
                                else current_beam.tokens,
                                logprobs=current_beam.logprobs +
                                [logprobs],
                                cum_logprob=current_beam.cum_logprob +
                                logprob_obj.logprob,
                                finish_reason="stop",
                                stop_reason=tokenizer.eos_token_id))
                    else:
                        new_beams.append(
                            BeamSearchSequence(
                                tokens=current_beam.tokens + [token_id],
                                logprobs=current_beam.logprobs +
                                [logprobs],
                                lora_request=current_beam.lora_request,
                                cum_logprob=current_beam.cum_logprob +
                                logprob_obj.logprob,
                                multi_modal_data=current_beam.
                                multi_modal_data,
                                mm_processor_kwargs=current_beam.
                                mm_processor_kwargs))

        sorted_beams = sorted(new_beams, key=sort_beams_key, reverse=True)
        all_beams = sorted_beams[:beam_width]

    completed.extend(all_beams)
    sorted_completed = sorted(completed, key=sort_beams_key, reverse=True)
    best_beams = sorted_completed[:beam_width]

    for beam in best_beams:
        if (beam.tokens[-1] == tokenizer.eos_token_id and not ignore_eos):
            # Skip the eos token in the text.
            tokens = beam.tokens[tokenized_length:-1]
        else:
            tokens = beam.tokens[tokenized_length:]
        beam.text = tokenizer.decode(tokens)

    beam_search_output = RequestOutput(
        request_id=request_id,
        prompt=prompt_text,
        outputs=[
            CompletionOutput(text=beam.text,
                             cumulative_logprob=beam.cum_logprob,
                             token_ids=beam.tokens[tokenized_length:],
                             index=i,
                             logprobs=beam.logprobs,
                             finish_reason=beam.finish_reason if
                             beam.finish_reason is not None else "length",
                             stop_reason=beam.stop_reason)
            for (i, beam) in enumerate(best_beams)
        ],
        finished=True,
        prompt_token_ids=prompt_token_ids,
        prompt_logprobs=None)

    yield beam_search_output

check_health abstractmethod async 

check_health() -> None

Raise if unhealthy

Source code in vllm/engine/protocol.py 
282
283
284
285
@abstractmethod
async def check_health(self) -> None:
    """Raise if unhealthy"""
    ...

do_log_stats abstractmethod async 

do_log_stats(
    scheduler_outputs: Optional[SchedulerOutputs] = None,
    model_output: Optional[list[SamplerOutput]] = None,
) -> None
Source code in vllm/engine/protocol.py 
274
275
276
277
278
279
280
@abstractmethod
async def do_log_stats(
    self,
    scheduler_outputs: Optional[SchedulerOutputs] = None,
    model_output: Optional[list[SamplerOutput]] = None,
) -> None:
    ...

encode abstractmethod 

encode(
    prompt: PromptType,
    pooling_params: PoolingParams,
    request_id: str,
    lora_request: Optional[LoRARequest] = None,
    trace_headers: Optional[Mapping[str, str]] = None,
    priority: int = 0,
) -> AsyncGenerator[PoolingRequestOutput, None]

Generate outputs for a request from a pooling model.

Source code in vllm/engine/protocol.py 
220
221
222
223
224
225
226
227
228
229
230
231
@abstractmethod
def encode(
    self,
    prompt: PromptType,
    pooling_params: PoolingParams,
    request_id: str,
    lora_request: Optional[LoRARequest] = None,
    trace_headers: Optional[Mapping[str, str]] = None,
    priority: int = 0,
) -> AsyncGenerator[PoolingRequestOutput, None]:
    """Generate outputs for a request from a pooling model."""
    ...

generate abstractmethod 

generate(
    prompt: PromptType,
    sampling_params: SamplingParams,
    request_id: str,
    lora_request: Optional[LoRARequest] = None,
    trace_headers: Optional[Mapping[str, str]] = None,
    prompt_adapter_request: Optional[
        PromptAdapterRequest
    ] = None,
    priority: int = 0,
) -> AsyncGenerator[RequestOutput, None]

Generate outputs for a request.

Source code in vllm/engine/protocol.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
@abstractmethod
def generate(
    self,
    prompt: PromptType,
    sampling_params: SamplingParams,
    request_id: str,
    lora_request: Optional[LoRARequest] = None,
    trace_headers: Optional[Mapping[str, str]] = None,
    prompt_adapter_request: Optional[PromptAdapterRequest] = None,
    priority: int = 0,
) -> AsyncGenerator[RequestOutput, None]:
    """Generate outputs for a request."""
    ...

get_decoding_config abstractmethod async 

get_decoding_config() -> DecodingConfig

Get the decoding configuration of the vLLM engine.

Source code in vllm/engine/protocol.py 
252
253
254
255
@abstractmethod
async def get_decoding_config(self) -> DecodingConfig:
    """Get the decoding configuration of the vLLM engine."""
    ...

get_input_preprocessor abstractmethod async 

get_input_preprocessor() -> InputPreprocessor

Get the input processor of the vLLM engine.

Source code in vllm/engine/protocol.py 
257
258
259
260
@abstractmethod
async def get_input_preprocessor(self) -> InputPreprocessor:
    """Get the input processor of the vLLM engine."""
    ...

get_model_config abstractmethod async 

get_model_config() -> ModelConfig

Get the model configuration of the vLLM engine.

Source code in vllm/engine/protocol.py 
247
248
249
250
@abstractmethod
async def get_model_config(self) -> ModelConfig:
    """Get the model configuration of the vLLM engine."""
    ...

get_tokenizer abstractmethod async 

get_tokenizer(
    lora_request: Optional[LoRARequest] = None,
) -> AnyTokenizer

Get the appropriate tokenizer for the request

Source code in vllm/engine/protocol.py 
262
263
264
265
266
267
268
@abstractmethod
async def get_tokenizer(
    self,
    lora_request: Optional[LoRARequest] = None,
) -> AnyTokenizer:
    """Get the appropriate tokenizer for the request"""
    ...

get_vllm_config abstractmethod async 

get_vllm_config() -> VllmConfig

Get the vllm configuration of the vLLM engine.

Source code in vllm/engine/protocol.py 
242
243
244
245
@abstractmethod
async def get_vllm_config(self) -> VllmConfig:
    """Get the vllm configuration of the vLLM engine."""
    ...

is_sleeping abstractmethod async 

is_sleeping() -> bool

Check whether the engine is sleeping

Source code in vllm/engine/protocol.py 
318
319
320
321
@abstractmethod
async def is_sleeping(self) -> bool:
    """Check whether the engine is sleeping"""
    ...

is_tracing_enabled abstractmethod async 

is_tracing_enabled() -> bool
Source code in vllm/engine/protocol.py 
270
271
272
@abstractmethod
async def is_tracing_enabled(self) -> bool:
    ...

reset_mm_cache abstractmethod async 

reset_mm_cache() -> None

Reset the multi-modal cache

Source code in vllm/engine/protocol.py 
297
298
299
300
@abstractmethod
async def reset_mm_cache(self) -> None:
    """Reset the multi-modal cache"""
    ...

reset_prefix_cache abstractmethod async 

reset_prefix_cache(device: Optional[Device] = None) -> None

Reset the prefix cache

Source code in vllm/engine/protocol.py 
302
303
304
305
306
@abstractmethod
async def reset_prefix_cache(self,
                             device: Optional[Device] = None) -> None:
    """Reset the prefix cache"""
    ...

sleep abstractmethod async 

sleep(level: int = 1) -> None

Sleep the engine

Source code in vllm/engine/protocol.py 
308
309
310
311
@abstractmethod
async def sleep(self, level: int = 1) -> None:
    """Sleep the engine"""
    ...

start_profile abstractmethod async 

start_profile() -> None

Start profiling the engine

Source code in vllm/engine/protocol.py 
287
288
289
290
@abstractmethod
async def start_profile(self) -> None:
    """Start profiling the engine"""
    ...

stop_profile abstractmethod async 

stop_profile() -> None

Start profiling the engine

Source code in vllm/engine/protocol.py 
292
293
294
295
@abstractmethod
async def stop_profile(self) -> None:
    """Start profiling the engine"""
    ...

wake_up abstractmethod async 

wake_up(tags: Optional[list[str]] = None) -> None

Wake up the engine

Source code in vllm/engine/protocol.py 
313
314
315
316
@abstractmethod
async def wake_up(self, tags: Optional[list[str]] = None) -> None:
    """Wake up the engine"""
    ...