| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | from typing import TYPE_CHECKING |
| 4 | |
| 5 | from vllm.entrypoints.anthropic.protocol import AnthropicMessagesRequest |
| 6 | from vllm.entrypoints.anthropic.serving import AnthropicServingMessages |
| 7 | from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest |
| 8 | from vllm.entrypoints.openai.completion.protocol import CompletionRequest |
| 9 | from vllm.entrypoints.openai.models.serving import ( |
| 10 | OpenAIModelRegistry, |
| 11 | OpenAIServingModels, |
| 12 | ) |
| 13 | from vllm.entrypoints.openai.responses.protocol import ResponsesRequest |
| 14 | from vllm.entrypoints.scale_out.token_in_token_out.mm_features import ( |
| 15 | extract_mm_features, |
| 16 | ) |
| 17 | from vllm.entrypoints.scale_out.token_in_token_out.protocol import ( |
| 18 | GenerateRequest, |
| 19 | MultiModalFeatures, |
| 20 | ) |
| 21 | from vllm.entrypoints.serve.engine.protocol import ErrorResponse |
| 22 | from vllm.entrypoints.serve.engine.serving import BaseServing |
| 23 | from vllm.entrypoints.serve.utils.api_utils import get_max_tokens |
| 24 | from vllm.entrypoints.serve.utils.request_logger import RequestLogger |
| 25 | from vllm.inputs import EngineInput |
| 26 | from vllm.logger import init_logger |
| 27 | from vllm.multimodal.parse import MultiModalDataParser |
| 28 | from vllm.renderers.inputs.preprocess import ( |
| 29 | extract_prompt_components, |
| 30 | extract_prompt_len, |
| 31 | ) |
| 32 | from vllm.renderers.online_renderer import OnlineRenderer |
| 33 | from vllm.utils import random_uuid |
| 34 | |
| 35 | logger = init_logger(__name__) |
| 36 | |
| 37 | if TYPE_CHECKING: |
| 38 | from vllm.entrypoints.mcp.tool_server import ToolServer |
| 39 | |
| 40 | |
| 41 | class ServingRender(BaseServing): |
| 42 | def __init__( |
| 43 | self, |
| 44 | models: OpenAIServingModels | OpenAIModelRegistry, |
| 45 | online_renderer: "OnlineRenderer", |
| 46 | *, |
| 47 | request_logger: RequestLogger | None = None, |
| 48 | tool_server: "ToolServer | None" = None, |
| 49 | ) -> None: |
| 50 | super().__init__( |
| 51 | models=models, |
| 52 | model_config=online_renderer.model_config, |
| 53 | request_logger=request_logger, |
| 54 | ) |
| 55 | |
| 56 | self.online_renderer = online_renderer |
| 57 | self.tool_server = tool_server |
| 58 | |
| 59 | self._merge_inline_system = ( |
| 60 | AnthropicServingMessages._should_merge_inline_system(online_renderer) |
| 61 | ) |
| 62 | |
| 63 | self._placeholder_metadata_parser: MultiModalDataParser | None = None |
| 64 | self._placeholder_metadata_parser_failed = False |
| 65 | |
| 66 | self.default_sampling_params = ( |
| 67 | online_renderer.model_config.get_diff_sampling_param() |
| 68 | ) |
| 69 | mc = online_renderer.model_config |
| 70 | self.override_max_tokens = ( |
| 71 | self.default_sampling_params.get("max_tokens") |
| 72 | if mc.generation_config not in ("auto", "vllm") |
| 73 | else getattr(mc, "override_generation_config", {}).get("max_new_tokens") |
| 74 | ) |
| 75 | |
| 76 | async def render_chat_request( |
| 77 | self, |
| 78 | request: ChatCompletionRequest, |
| 79 | ) -> GenerateRequest | ErrorResponse: |
| 80 | """Validate the model and preprocess a chat completion request. |
| 81 | |
| 82 | This is the authoritative implementation used directly by the |
| 83 | GPU-less render server and delegated to by OpenAIServingChat. |
| 84 | """ |
| 85 | error_check_ret = await self._check_model(request) |
| 86 | if error_check_ret is not None: |
| 87 | logger.error("Error with model %s", error_check_ret) |
| 88 | return error_check_ret |
| 89 | |
| 90 | if request.use_beam_search: |
| 91 | return self.create_error_response( |
| 92 | "Beam search is not supported by the render endpoint" |
| 93 | ) |
| 94 | |
| 95 | result = await self.online_renderer.render_chat(request, skip_mm_cache=True) |
| 96 | if isinstance(result, ErrorResponse): |
| 97 | return result |
| 98 | |
| 99 | _, engine_inputs = result |
| 100 | |
| 101 | if len(engine_inputs) != 1: |
| 102 | return self.create_error_response( |
| 103 | f"Expected exactly 1 engine prompt, got {len(engine_inputs)}" |
| 104 | ) |
| 105 | |
| 106 | engine_input = engine_inputs[0] |
| 107 | |
| 108 | prompt_components = extract_prompt_components(self.model_config, engine_input) |
| 109 | token_ids = prompt_components.token_ids |
| 110 | if not token_ids: |
| 111 | return self.create_error_response("No token_ids rendered") |
| 112 | token_ids = list(token_ids) |
| 113 | |
| 114 | input_length = extract_prompt_len(self.model_config, engine_input) |
| 115 | max_tokens = get_max_tokens( |
| 116 | self.model_config.max_model_len, |
| 117 | request.max_completion_tokens |
| 118 | if request.max_completion_tokens is not None |
| 119 | else request.max_tokens, |
| 120 | input_length, |
| 121 | self.default_sampling_params, |
| 122 | self.override_max_tokens, |
| 123 | truncate_prompt_tokens=request.truncate_prompt_tokens, |
| 124 | ) |
| 125 | params = request.to_sampling_params(max_tokens, self.default_sampling_params) |
| 126 | |
| 127 | request_id = f"chatcmpl-{random_uuid()}" |
| 128 | |
| 129 | return GenerateRequest( |
| 130 | request_id=request_id, |
| 131 | token_ids=token_ids, |
| 132 | features=self._extract_mm_features(engine_input), |
| 133 | sampling_params=params, |
| 134 | model=request.model, |
| 135 | stream=bool(request.stream), |
| 136 | stream_options=(request.stream_options if request.stream else None), |
| 137 | cache_salt=request.cache_salt, |
| 138 | priority=request.priority, |
| 139 | token_offsets=engine_input.get("prompt_token_offsets"), |
| 140 | ) |
| 141 | |
| 142 | async def render_messages_request( |
| 143 | self, |
| 144 | request: AnthropicMessagesRequest, |
| 145 | ) -> GenerateRequest | ErrorResponse: |
| 146 | """Validate the model and preprocess an Anthropic Messages request. |
| 147 | |
| 148 | Converts the request to the OpenAI chat format using the same |
| 149 | conversion as the /v1/messages server path, then delegates to |
| 150 | render_chat_request so the rendered tokens match the server exactly. |
| 151 | """ |
| 152 | chat_req = AnthropicServingMessages.to_chat_completion_request( |
| 153 | request, merge_inline_system=self._merge_inline_system |
| 154 | ) |
| 155 | return await self.render_chat_request(chat_req) |
| 156 | |
| 157 | async def render_completion_request( |
| 158 | self, |
| 159 | request: CompletionRequest, |
| 160 | ) -> list[GenerateRequest] | ErrorResponse: |
| 161 | """Validate the model and preprocess a completion request. |
| 162 | |
| 163 | This is the authoritative implementation used directly by the |
| 164 | GPU-less render server and delegated to by OpenAIServingCompletion. |
| 165 | """ |
| 166 | error_check_ret = await self._check_model(request) |
| 167 | if error_check_ret is not None: |
| 168 | return error_check_ret |
| 169 | result = await self.online_renderer.render_completion( |
| 170 | request, skip_mm_cache=True |
| 171 | ) |
| 172 | if isinstance(result, ErrorResponse): |
| 173 | return result |
| 174 | generate_requests: list[GenerateRequest] = [] |
| 175 | for engine_input in result: |
| 176 | prompt_components = extract_prompt_components( |
| 177 | self.model_config, engine_input |
| 178 | ) |
| 179 | token_ids = prompt_components.token_ids |
| 180 | if not token_ids: |
| 181 | return self.create_error_response("No token_ids rendered") |
| 182 | token_ids = list(token_ids) |
| 183 | |
| 184 | input_length = extract_prompt_len(self.model_config, engine_input) |
| 185 | max_tokens = get_max_tokens( |
| 186 | self.model_config.max_model_len, |
| 187 | request.max_tokens, |
| 188 | input_length, |
| 189 | self.default_sampling_params, |
| 190 | self.override_max_tokens, |
| 191 | truncate_prompt_tokens=request.truncate_prompt_tokens, |
| 192 | ) |
| 193 | params = request.to_sampling_params( |
| 194 | max_tokens, self.default_sampling_params |
| 195 | ) |
| 196 | |
| 197 | request_id = f"cmpl-{random_uuid()}" |
| 198 | |
| 199 | generate_requests.append( |
| 200 | GenerateRequest( |
| 201 | request_id=request_id, |
| 202 | token_ids=token_ids, |
| 203 | features=self._extract_mm_features(engine_input), |
| 204 | sampling_params=params, |
| 205 | model=request.model, |
| 206 | stream=bool(request.stream), |
| 207 | stream_options=(request.stream_options if request.stream else None), |
| 208 | cache_salt=request.cache_salt, |
| 209 | priority=request.priority, |
| 210 | token_offsets=engine_input.get("prompt_token_offsets"), |
| 211 | ) |
| 212 | ) |
| 213 | |
| 214 | return generate_requests |
| 215 | |
| 216 | async def render_responses_request( |
| 217 | self, |
| 218 | request: ResponsesRequest, |
| 219 | ) -> GenerateRequest | ErrorResponse: |
| 220 | error_check_ret = await self._check_model(request) |
| 221 | if error_check_ret is not None: |
| 222 | return error_check_ret |
| 223 | if request.previous_response_id is not None: |
| 224 | return self.create_error_response( |
| 225 | message=( |
| 226 | "previous_response_id is not supported by the stateless " |
| 227 | "render endpoint." |
| 228 | ), |
| 229 | err_type="invalid_request_error", |
| 230 | param="previous_response_id", |
| 231 | ) |
| 232 | |
| 233 | result = await self.online_renderer.render_responses( |
| 234 | request, |
| 235 | previous_messages=None, |
| 236 | previous_response_outputs=None, |
| 237 | tool_server=self.tool_server, |
| 238 | skip_mm_cache=True, |
| 239 | ) |
| 240 | if isinstance(result, ErrorResponse): |
| 241 | return result |
| 242 | |
| 243 | engine_input = result.engine_input |
| 244 | prompt_components = extract_prompt_components(self.model_config, engine_input) |
| 245 | token_ids = prompt_components.token_ids |
| 246 | if not token_ids: |
| 247 | return self.create_error_response("No token_ids rendered") |
| 248 | |
| 249 | input_length = extract_prompt_len(self.model_config, engine_input) |
| 250 | max_tokens = get_max_tokens( |
| 251 | self.model_config.max_model_len, |
| 252 | request.max_output_tokens, |
| 253 | input_length, |
| 254 | self.default_sampling_params, |
| 255 | self.override_max_tokens, |
| 256 | truncate_prompt_tokens=(-1 if request.truncation != "disabled" else None), |
| 257 | ) |
| 258 | params = request.to_sampling_params(max_tokens, self.default_sampling_params) |
| 259 | |
| 260 | return GenerateRequest( |
| 261 | request_id=request.request_id, |
| 262 | token_ids=list(token_ids), |
| 263 | features=self._extract_mm_features(engine_input), |
| 264 | sampling_params=params, |
| 265 | model=request.model, |
| 266 | stream=bool(request.stream), |
| 267 | cache_salt=request.cache_salt, |
| 268 | priority=request.priority, |
| 269 | kv_transfer_params=request.kv_transfer_params, |
| 270 | ec_transfer_params=request.ec_transfer_params, |
| 271 | token_offsets=engine_input.get("prompt_token_offsets"), |
| 272 | ) |
| 273 | |
| 274 | def _placeholder_metadata_fields(self, modality: str) -> set[str]: |
| 275 | """Fields the EC consumer still requires after embeddings transfer.""" |
| 276 | if self._placeholder_metadata_parser_failed: |
| 277 | return set() |
| 278 | if self._placeholder_metadata_parser is None: |
| 279 | try: |
| 280 | from vllm.multimodal import MULTIMODAL_REGISTRY |
| 281 | |
| 282 | self._placeholder_metadata_parser = ( |
| 283 | MULTIMODAL_REGISTRY.create_processor( |
| 284 | self.model_config |
| 285 | ).info.data_parser |
| 286 | ) |
| 287 | except Exception: |
| 288 | logger.debug( |
| 289 | "Could not load placeholder metadata fields; " |
| 290 | "mm_metadata will use keep_on_cpu fields only." |
| 291 | ) |
| 292 | self._placeholder_metadata_parser_failed = True |
| 293 | return set() |
| 294 | return set( |
| 295 | self._placeholder_metadata_parser.placeholder_metadata_fields(modality) |
| 296 | ) |
| 297 | |
| 298 | def _extract_mm_features( |
| 299 | self, |
| 300 | engine_input: EngineInput, |
| 301 | ) -> MultiModalFeatures | None: |
| 302 | return extract_mm_features( |
| 303 | engine_input, |
| 304 | metadata_fields_for=self._placeholder_metadata_fields, |
| 305 | ) |