| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | import json |
| 4 | import time |
| 5 | from collections.abc import Awaitable, Mapping |
| 6 | from dataclasses import dataclass, field |
| 7 | from http import HTTPStatus |
| 8 | from typing import ClassVar, Generic, TypeVar |
| 9 | |
| 10 | from fastapi import Request |
| 11 | from pydantic import ConfigDict |
| 12 | from starlette.datastructures import Headers |
| 13 | |
| 14 | from vllm import RequestOutput |
| 15 | from vllm.engine.protocol import EngineClient |
| 16 | from vllm.entrypoints.generate.base.protocol import ( |
| 17 | PerRequestMetrics, |
| 18 | SpeculativeDecodingMetrics, |
| 19 | ) |
| 20 | from vllm.entrypoints.generate.beam_search.online import BeamSearchOnlineMixin |
| 21 | from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest |
| 22 | from vllm.entrypoints.openai.completion.protocol import CompletionRequest |
| 23 | from vllm.entrypoints.openai.models.serving import OpenAIServingModels |
| 24 | from vllm.entrypoints.openai.responses.protocol import ResponsesRequest |
| 25 | from vllm.entrypoints.serve.engine.protocol import ErrorResponse |
| 26 | from vllm.entrypoints.serve.engine.serving import BaseServing |
| 27 | from vllm.entrypoints.serve.engine.typing import AnyRequest |
| 28 | from vllm.entrypoints.serve.utils.request_logger import RequestLogger |
| 29 | from vllm.exceptions import GenerationError |
| 30 | from vllm.inputs import EngineInput |
| 31 | from vllm.logger import init_logger |
| 32 | from vllm.logprobs import Logprob, PromptLogprobs |
| 33 | from vllm.lora.request import LoRARequest |
| 34 | from vllm.tokenizers import TokenizerLike |
| 35 | from vllm.tracing import ( |
| 36 | contains_trace_headers, |
| 37 | extract_trace_headers, |
| 38 | log_tracing_disabled_warning, |
| 39 | ) |
| 40 | from vllm.v1.metrics.stats import RequestStateStats |
| 41 | |
| 42 | logger = init_logger(__name__) |
| 43 | |
| 44 | RequestT = TypeVar("RequestT", bound=AnyRequest) |
| 45 | _T = TypeVar("_T") |
| 46 | SESSION_ID_HEADER = "X-Session-ID" |
| 47 | PRIORITY_HEADER = "X-Vllm-Priority" |
| 48 | |
| 49 | |
| 50 | def build_per_request_timing_metrics( |
| 51 | metrics: RequestStateStats | None, |
| 52 | num_generation_tokens: int, |
| 53 | ) -> PerRequestMetrics: |
| 54 | """Build per-request timing metrics from ``RequestStateStats``. |
| 55 | |
| 56 | ``generation_time_ms`` is the decode interval only (first output token to |
| 57 | last output token); it excludes both queue wait and prefill/TTFT. |
| 58 | ``tokens_per_second`` is overall output throughput: all generated tokens |
| 59 | over the inference interval (scheduling to last output token), so it counts |
| 60 | the prefill/TTFT phase and is not simply the reciprocal of ``mean_itl_ms``. |
| 61 | Each field is left ``None`` when the timestamps it depends on are |
| 62 | unavailable. |
| 63 | """ |
| 64 | if metrics is None: |
| 65 | return PerRequestMetrics() |
| 66 | |
| 67 | queued_ts = metrics.queued_ts |
| 68 | scheduled_ts = metrics.scheduled_ts |
| 69 | first_token_ts = metrics.first_token_ts |
| 70 | last_token_ts = metrics.last_token_ts |
| 71 | |
| 72 | time_to_first_token_ms: float | None = None |
| 73 | generation_time_ms: float | None = None |
| 74 | queue_time_ms: float | None = None |
| 75 | mean_itl_ms: float | None = None |
| 76 | tokens_per_second: float | None = None |
| 77 | |
| 78 | if scheduled_ts > 0 and first_token_ts > 0: |
| 79 | time_to_first_token_ms = (first_token_ts - scheduled_ts) * 1000 |
| 80 | |
| 81 | if first_token_ts > 0 and last_token_ts > 0: |
| 82 | generation_time_ms = (last_token_ts - first_token_ts) * 1000 |
| 83 | |
| 84 | if queued_ts > 0 and scheduled_ts > 0: |
| 85 | queue_time_ms = (scheduled_ts - queued_ts) * 1000 |
| 86 | |
| 87 | if first_token_ts > 0 and last_token_ts > 0 and num_generation_tokens > 1: |
| 88 | decode_time = last_token_ts - first_token_ts |
| 89 | mean_itl_ms = decode_time / (num_generation_tokens - 1) * 1000 |
| 90 | |
| 91 | if scheduled_ts > 0 and last_token_ts > 0: |
| 92 | inference_time_ms = (last_token_ts - scheduled_ts) * 1000 |
| 93 | if inference_time_ms > 0: |
| 94 | tokens_per_second = num_generation_tokens / inference_time_ms * 1000 |
| 95 | |
| 96 | return PerRequestMetrics( |
| 97 | time_to_first_token_ms=time_to_first_token_ms, |
| 98 | generation_time_ms=generation_time_ms, |
| 99 | queue_time_ms=queue_time_ms, |
| 100 | mean_itl_ms=mean_itl_ms, |
| 101 | tokens_per_second=tokens_per_second, |
| 102 | ) |
| 103 | |
| 104 | |
| 105 | def build_spec_decoding_metrics( |
| 106 | final_res: "RequestOutput | None", |
| 107 | ) -> SpeculativeDecodingMetrics | None: |
| 108 | """Build per-request spec-decode acceptance metrics from the single output |
| 109 | sequence, or ``None`` when unavailable (metrics disabled, or no sequence |
| 110 | yet). |
| 111 | |
| 112 | Only meaningful for single-sequence requests; callers suppress it for n>1. |
| 113 | """ |
| 114 | if final_res is None or not final_res.outputs: |
| 115 | return None |
| 116 | metrics = final_res.outputs[0].spec_decode_metrics |
| 117 | if metrics is None: |
| 118 | return None |
| 119 | return SpeculativeDecodingMetrics(**metrics.to_dict()) |
| 120 | |
| 121 | |
| 122 | @dataclass(kw_only=True) |
| 123 | class ServeContext(Generic[RequestT]): |
| 124 | request: RequestT |
| 125 | raw_request: Request | None = None |
| 126 | model_name: str |
| 127 | request_id: str |
| 128 | created_time: int = field(default_factory=lambda: int(time.time())) |
| 129 | lora_request: LoRARequest | None = None |
| 130 | engine_inputs: list[EngineInput] | None = None |
| 131 | model_config = ConfigDict(arbitrary_types_allowed=True) |
| 132 | |
| 133 | |
| 134 | class GenerateBaseServing(BaseServing, BeamSearchOnlineMixin): |
| 135 | request_id_prefix: ClassVar[str] = """ |
| 136 | A short string prepended to every request’s ID. |
| 137 | """ |
| 138 | |
| 139 | def __init__( |
| 140 | self, |
| 141 | engine_client: EngineClient, |
| 142 | models: OpenAIServingModels, |
| 143 | *, |
| 144 | request_logger: RequestLogger | None, |
| 145 | return_tokens_as_token_ids: bool = False, |
| 146 | ): |
| 147 | super().__init__( |
| 148 | models=models, |
| 149 | model_config=engine_client.model_config, |
| 150 | request_logger=request_logger, |
| 151 | ) |
| 152 | |
| 153 | self.engine_client = engine_client |
| 154 | self.return_tokens_as_token_ids = return_tokens_as_token_ids |
| 155 | self.renderer = engine_client.renderer |
| 156 | self.input_processor = engine_client.input_processor |
| 157 | vllm_config = getattr(engine_client, "vllm_config", None) |
| 158 | kv_transfer_config = getattr(vllm_config, "kv_transfer_config", None) |
| 159 | self.has_kv_connector = kv_transfer_config is not None |
| 160 | |
| 161 | # Computed once at startup (cached by ``vllm_config`` identity) and |
| 162 | # stamped on non-streaming responses. Streaming chunks deliberately |
| 163 | # omit it to avoid per-chunk overhead. |
| 164 | from vllm.entrypoints.serve.utils.fingerprint import get_system_fingerprint |
| 165 | |
| 166 | try: |
| 167 | self.system_fingerprint: str | None = get_system_fingerprint( |
| 168 | engine_client.vllm_config |
| 169 | ) |
| 170 | except Exception: |
| 171 | # Never fail server startup over the fingerprint. |
| 172 | self.system_fingerprint = None |
| 173 | |
| 174 | def _preflight(self, n: int = 1) -> None: |
| 175 | """Engine checks that must run before a response is started. |
| 176 | |
| 177 | This is required for the streaming case, where we return a success |
| 178 | status before we actually start generating text :). |
| 179 | |
| 180 | Args: |
| 181 | n: Number of sequences the request will occupy. |
| 182 | |
| 183 | """ |
| 184 | if self.engine_client.errored: |
| 185 | raise self.engine_client.dead_error |
| 186 | self.engine_client.check_admission(n) |
| 187 | |
| 188 | def create_streaming_error_response( |
| 189 | self, |
| 190 | message: str | Exception, |
| 191 | err_type: str = "BadRequestError", |
| 192 | status_code: HTTPStatus = HTTPStatus.BAD_REQUEST, |
| 193 | param: str | None = None, |
| 194 | ) -> str: |
| 195 | json_str = json.dumps( |
| 196 | self.create_error_response( |
| 197 | message=message, |
| 198 | err_type=err_type, |
| 199 | status_code=status_code, |
| 200 | param=param, |
| 201 | ).model_dump() |
| 202 | ) |
| 203 | return json_str |
| 204 | |
| 205 | def _raise_if_error(self, finish_reason: str | None, request_id: str) -> None: |
| 206 | """Raise GenerationError if finish_reason indicates an error.""" |
| 207 | if finish_reason == "error": |
| 208 | logger.error( |
| 209 | "Request %s failed with an internal error during generation", |
| 210 | request_id, |
| 211 | ) |
| 212 | raise GenerationError("Internal server error") |
| 213 | |
| 214 | def _convert_generation_error_to_streaming_response( |
| 215 | self, e: GenerationError |
| 216 | ) -> str: |
| 217 | """Convert GenerationError to streaming error response.""" |
| 218 | return self.create_streaming_error_response( |
| 219 | str(e), |
| 220 | err_type="InternalServerError", |
| 221 | status_code=e.status_code, |
| 222 | ) |
| 223 | |
| 224 | async def _get_trace_headers( |
| 225 | self, |
| 226 | headers: Headers, |
| 227 | ) -> Mapping[str, str] | None: |
| 228 | is_tracing_enabled = await self.engine_client.is_tracing_enabled() |
| 229 | |
| 230 | if is_tracing_enabled: |
| 231 | return extract_trace_headers(headers) |
| 232 | |
| 233 | if contains_trace_headers(headers): |
| 234 | log_tracing_disabled_warning() |
| 235 | |
| 236 | return None |
| 237 | |
| 238 | @staticmethod |
| 239 | def _get_data_parallel_rank(raw_request: Request | None) -> int | None: |
| 240 | """Pulls the data parallel rank from a header, if provided.""" |
| 241 | if raw_request is None: |
| 242 | return None |
| 243 | |
| 244 | rank_str = raw_request.headers.get("X-data-parallel-rank") |
| 245 | if rank_str is None: |
| 246 | return None |
| 247 | |
| 248 | try: |
| 249 | return int(rank_str) |
| 250 | except ValueError: |
| 251 | return None |
| 252 | |
| 253 | @staticmethod |
| 254 | def _get_session_id_from_headers(raw_request: Request | None) -> str | None: |
| 255 | if raw_request is None: |
| 256 | return None |
| 257 | if value := raw_request.headers.get(SESSION_ID_HEADER): |
| 258 | return value |
| 259 | return None |
| 260 | |
| 261 | @staticmethod |
| 262 | def _get_session_id( |
| 263 | request: ChatCompletionRequest | CompletionRequest | ResponsesRequest, |
| 264 | raw_request: Request | None, |
| 265 | ) -> str | None: |
| 266 | if request.session_id: |
| 267 | return request.session_id |
| 268 | if value := GenerateBaseServing._get_session_id_from_headers(raw_request): |
| 269 | return value |
| 270 | if request.vllm_xargs: |
| 271 | session_id = request.vllm_xargs.get("session_id") |
| 272 | if isinstance(session_id, str) and session_id: |
| 273 | return session_id |
| 274 | return None |
| 275 | |
| 276 | @staticmethod |
| 277 | def _get_priority( |
| 278 | request: ChatCompletionRequest | CompletionRequest | ResponsesRequest, |
| 279 | raw_request: Request | None, |
| 280 | ) -> int: |
| 281 | if raw_request is not None: |
| 282 | priority = raw_request.headers.get(PRIORITY_HEADER) |
| 283 | if priority is not None: |
| 284 | try: |
| 285 | return int(priority) |
| 286 | except ValueError: |
| 287 | pass |
| 288 | return request.priority |
| 289 | |
| 290 | async def _with_kv_transfer_rejection_cleanup( |
| 291 | self, |
| 292 | awaitable: Awaitable[_T], |
| 293 | request: ChatCompletionRequest | CompletionRequest | ResponsesRequest, |
| 294 | raw_request: Request | None, |
| 295 | ) -> _T: |
| 296 | """Wrap a `create_*` coroutine so that, if it raises or returns an |
| 297 | ErrorResponse (i.e. the request never reached the engine), the KV |
| 298 | connector is notified to free any pinned remote-prefill blocks.""" |
| 299 | kv_transfer_params = self.has_kv_connector and request.kv_transfer_params |
| 300 | if not kv_transfer_params or not kv_transfer_params.get("do_remote_prefill"): |
| 301 | return await awaitable |
| 302 | |
| 303 | notify = True |
| 304 | try: |
| 305 | result = await awaitable |
| 306 | if not isinstance(result, ErrorResponse): |
| 307 | notify = False |
| 308 | return result |
| 309 | finally: |
| 310 | if notify: |
| 311 | try: |
| 312 | await self.engine_client.notify_kv_transfer_request_rejected( |
| 313 | request.request_id, |
| 314 | kv_transfer_params, |
| 315 | data_parallel_rank=self._get_data_parallel_rank(raw_request), |
| 316 | ) |
| 317 | except Exception: |
| 318 | logger.warning( |
| 319 | "Failed to notify KV connector about rejected request %s", |
| 320 | request.request_id, |
| 321 | exc_info=True, |
| 322 | ) |
| 323 | |
| 324 | @staticmethod |
| 325 | def _get_decoded_token( |
| 326 | logprob: Logprob, |
| 327 | token_id: int, |
| 328 | tokenizer: TokenizerLike | None, |
| 329 | return_as_token_id: bool = False, |
| 330 | ) -> str: |
| 331 | if return_as_token_id: |
| 332 | return format_token_id_placeholder(token_id) |
| 333 | |
| 334 | if logprob.decoded_token is not None: |
| 335 | return logprob.decoded_token |
| 336 | |
| 337 | if tokenizer is None: |
| 338 | raise ValueError( |
| 339 | "Unable to get tokenizer because `skip_tokenizer_init=True`" |
| 340 | ) |
| 341 | |
| 342 | return tokenizer.decode([token_id]) |
| 343 | |
| 344 | |
| 345 | def format_token_id_placeholder(token_id: int) -> str: |
| 346 | return f"token_id:{token_id}" |
| 347 | |
| 348 | |
| 349 | def resolve_token_id_placeholder( |
| 350 | token: str, tokenizer: TokenizerLike |
| 351 | ) -> tuple[str, list[int] | None]: |
| 352 | """Decode a 'token_id:N' placeholder back to a token string and UTF-8 bytes. |
| 353 | |
| 354 | Returns (token, None) unchanged if token is not a placeholder. |
| 355 | This is the inverse of format_token_id_placeholder / _get_decoded_token |
| 356 | when return_as_token_id=True. |
| 357 | """ |
| 358 | suffix = token.removeprefix("token_id:") |
| 359 | if suffix == token: |
| 360 | return token, None |
| 361 | try: |
| 362 | token_id = int(suffix) |
| 363 | except ValueError: |
| 364 | return token, None |
| 365 | token_repr = tokenizer.convert_ids_to_tokens([token_id])[0] |
| 366 | if token_repr is None: |
| 367 | logger.warning_once( |
| 368 | "resolve_token_id_placeholder: token_id %d has no vocab entry; " |
| 369 | "substituting empty string", |
| 370 | token_id, |
| 371 | ) |
| 372 | return "", None |
| 373 | token_str = tokenizer.convert_tokens_to_string([token_repr]) |
| 374 | return token_str, list(token_str.encode("utf-8", errors="replace")) |
| 375 | |
| 376 | |
| 377 | def clamp_prompt_logprobs( |
| 378 | prompt_logprobs: PromptLogprobs | None, |
| 379 | ) -> PromptLogprobs | None: |
| 380 | if prompt_logprobs is None: |
| 381 | return prompt_logprobs |
| 382 | |
| 383 | for logprob_dict in prompt_logprobs: |
| 384 | if logprob_dict is None: |
| 385 | continue |
| 386 | for logprob_values in logprob_dict.values(): |
| 387 | if logprob_values.logprob == float("-inf"): |
| 388 | logprob_values.logprob = -9999.0 |
| 389 | return prompt_logprobs |