| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | import time |
| 4 | |
| 5 | import vllm.envs as envs |
| 6 | from vllm.entrypoints.openai.chat_completion.protocol import ( |
| 7 | ChatCompletionResponse, |
| 8 | ChatCompletionStreamResponse, |
| 9 | ) |
| 10 | from vllm.entrypoints.openai.completion.protocol import ( |
| 11 | CompletionResponse, |
| 12 | CompletionStreamResponse, |
| 13 | ) |
| 14 | from vllm.entrypoints.openai.models.serving import ( |
| 15 | OpenAIModelRegistry, |
| 16 | OpenAIServingModels, |
| 17 | ) |
| 18 | from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo |
| 19 | from vllm.entrypoints.serve.engine.serving import BaseServing |
| 20 | from vllm.entrypoints.serve.utils.request_logger import RequestLogger |
| 21 | from vllm.inputs import EngineInput |
| 22 | from vllm.logger import init_logger |
| 23 | from vllm.renderers.online_derenderer import OnlineDerenderer |
| 24 | |
| 25 | from ..token_in_token_out.mm_features import extract_mm_features |
| 26 | from ..token_in_token_out.protocol import ( |
| 27 | DerenderChatRequest, |
| 28 | DerenderChatStreamRequest, |
| 29 | DerenderCompletionRequest, |
| 30 | DerenderCompletionStreamRequest, |
| 31 | DerenderStreamState, |
| 32 | GenerateResponse, |
| 33 | MultiModalFeatures, |
| 34 | ) |
| 35 | |
| 36 | logger = init_logger(__name__) |
| 37 | |
| 38 | |
| 39 | class ServingDerender(BaseServing): |
| 40 | def __init__( |
| 41 | self, |
| 42 | models: OpenAIServingModels | OpenAIModelRegistry, |
| 43 | online_derenderer: "OnlineDerenderer", |
| 44 | *, |
| 45 | request_logger: RequestLogger | None = None, |
| 46 | ) -> None: |
| 47 | super().__init__( |
| 48 | models=models, |
| 49 | model_config=models.model_config, |
| 50 | request_logger=request_logger, |
| 51 | ) |
| 52 | |
| 53 | self.online_derenderer = online_derenderer |
| 54 | |
| 55 | def _validate_derender_bounds( |
| 56 | self, |
| 57 | generate_responses: list[GenerateResponse], |
| 58 | ) -> ErrorResponse | None: |
| 59 | """Reject derender payloads that exceed resource bounds. |
| 60 | |
| 61 | Runs before any tokenizer.decode() or parser invocation to prevent |
| 62 | CPU/memory exhaustion from oversized caller-supplied token structures. |
| 63 | """ |
| 64 | max_n = envs.VLLM_MAX_N_SEQUENCES |
| 65 | max_model_len = self.model_config.max_model_len |
| 66 | # See ModelConfig.max_logprobs for semantics and default value. |
| 67 | max_logprobs = self.model_config.max_logprobs |
| 68 | |
| 69 | if len(generate_responses) > max_n: |
| 70 | return self.create_error_response( |
| 71 | f"generate_responses count ({len(generate_responses)}) " |
| 72 | f"exceeds server maximum ({max_n}). " |
| 73 | f"Set VLLM_MAX_N_SEQUENCES to increase this limit." |
| 74 | ) |
| 75 | |
| 76 | for gen in generate_responses: |
| 77 | if len(gen.choices) > max_n: |
| 78 | return self.create_error_response( |
| 79 | f"choices count ({len(gen.choices)}) in response " |
| 80 | f"'{gen.request_id}' exceeds server maximum ({max_n})." |
| 81 | ) |
| 82 | |
| 83 | for choice in gen.choices: |
| 84 | if choice.token_ids and len(choice.token_ids) > max_model_len: |
| 85 | return self.create_error_response( |
| 86 | f"token_ids length ({len(choice.token_ids)}) in " |
| 87 | f"choice {choice.index} exceeds " |
| 88 | f"max_model_len ({max_model_len})." |
| 89 | ) |
| 90 | if choice.logprobs and choice.logprobs.content: |
| 91 | if len(choice.logprobs.content) > max_model_len: |
| 92 | return self.create_error_response( |
| 93 | f"logprobs.content length " |
| 94 | f"({len(choice.logprobs.content)}) in " |
| 95 | f"choice {choice.index} exceeds " |
| 96 | f"max_model_len ({max_model_len})." |
| 97 | ) |
| 98 | for entry in choice.logprobs.content: |
| 99 | if ( |
| 100 | max_logprobs >= 0 |
| 101 | and entry.top_logprobs |
| 102 | and len(entry.top_logprobs) > max_logprobs |
| 103 | ): |
| 104 | return self.create_error_response( |
| 105 | f"top_logprobs count " |
| 106 | f"({len(entry.top_logprobs)}) in " |
| 107 | f"choice {choice.index} exceeds " |
| 108 | f"max_logprobs ({max_logprobs})." |
| 109 | ) |
| 110 | |
| 111 | if gen.prompt_logprobs and len(gen.prompt_logprobs) > max_model_len: |
| 112 | return self.create_error_response( |
| 113 | f"prompt_logprobs length ({len(gen.prompt_logprobs)}) " |
| 114 | f"in response '{gen.request_id}' exceeds " |
| 115 | f"max_model_len ({max_model_len})." |
| 116 | ) |
| 117 | |
| 118 | return None |
| 119 | |
| 120 | def _validate_prompt_token_ids( |
| 121 | self, |
| 122 | prompt_token_ids: list[list[int] | None], |
| 123 | ) -> ErrorResponse | None: |
| 124 | """Reject caller-supplied prompt_token_ids longer than max_model_len.""" |
| 125 | max_model_len = self.model_config.max_model_len |
| 126 | for ids in prompt_token_ids: |
| 127 | if ids is not None and len(ids) > max_model_len: |
| 128 | return self.create_error_response( |
| 129 | f"prompt_token_ids length ({len(ids)}) exceeds " |
| 130 | f"max_model_len ({max_model_len})." |
| 131 | ) |
| 132 | return None |
| 133 | |
| 134 | async def derender_chat_response( |
| 135 | self, |
| 136 | request: DerenderChatRequest, |
| 137 | ) -> ChatCompletionResponse | ErrorResponse: |
| 138 | """Postprocess a GenerateResponse into a ChatCompletionResponse. |
| 139 | |
| 140 | Non-streaming only: expects the complete GenerateResponse with all |
| 141 | token IDs present. Uses ``parser.parse()`` for one-shot extraction. |
| 142 | |
| 143 | When ``request.chat_request`` is provided, the parser splits the |
| 144 | output into (reasoning, content, tool_calls). Otherwise falls |
| 145 | back to plain detokenization. |
| 146 | """ |
| 147 | error_check_ret = await self._check_model(request) |
| 148 | if error_check_ret is not None: |
| 149 | return error_check_ret |
| 150 | |
| 151 | bounds_error = self._validate_derender_bounds([request.generate_response]) |
| 152 | if bounds_error is not None: |
| 153 | return bounds_error |
| 154 | |
| 155 | prompt_ids_error = self._validate_prompt_token_ids([request.prompt_token_ids]) |
| 156 | if prompt_ids_error is not None: |
| 157 | return prompt_ids_error |
| 158 | |
| 159 | if self.online_derenderer.parser is not None and request.chat_request is None: |
| 160 | return self.create_error_response( |
| 161 | "chat_request is required when a tool or reasoning parser is " |
| 162 | "configured because plain detokenization would leak raw parser " |
| 163 | "markup into content." |
| 164 | ) |
| 165 | |
| 166 | try: |
| 167 | choices = await self.online_derenderer.derender_chat( |
| 168 | request.generate_response, |
| 169 | request.chat_request, |
| 170 | request.prompt_token_ids, |
| 171 | ) |
| 172 | except ValueError as exc: |
| 173 | return self.create_error_response(str(exc)) |
| 174 | |
| 175 | prompt_tokens = ( |
| 176 | request.prompt_tokens if request.prompt_tokens is not None else 0 |
| 177 | ) |
| 178 | gen = request.generate_response |
| 179 | completion_tokens = sum(len(ch.token_ids) for ch in gen.choices if ch.token_ids) |
| 180 | usage = UsageInfo( |
| 181 | prompt_tokens=prompt_tokens, |
| 182 | completion_tokens=completion_tokens, |
| 183 | total_tokens=prompt_tokens + completion_tokens, |
| 184 | ) |
| 185 | |
| 186 | model_name = request.model or self.models.model_name() |
| 187 | logger.debug( |
| 188 | "derender_chat request_id=%s model=%s choices=%d completion_tokens=%d", |
| 189 | gen.request_id, |
| 190 | model_name, |
| 191 | len(choices), |
| 192 | completion_tokens, |
| 193 | ) |
| 194 | return ChatCompletionResponse( |
| 195 | id=gen.request_id, |
| 196 | model=model_name, |
| 197 | created=int(time.time()), |
| 198 | choices=choices, |
| 199 | usage=usage, |
| 200 | prompt_logprobs=gen.prompt_logprobs, |
| 201 | kv_transfer_params=gen.kv_transfer_params, |
| 202 | metrics=gen.metrics, |
| 203 | ) |
| 204 | |
| 205 | async def derender_completion_response( |
| 206 | self, |
| 207 | request: DerenderCompletionRequest, |
| 208 | ) -> CompletionResponse | ErrorResponse: |
| 209 | """Postprocess a list of GenerateResponses into a CompletionResponse. |
| 210 | |
| 211 | Non-streaming only. Mirrors the multi-prompt completions case: one |
| 212 | GenerateResponse per prompt, parallel to the list[GenerateRequest] |
| 213 | from /v1/completions/render. |
| 214 | """ |
| 215 | error_check_ret = await self._check_model(request) |
| 216 | if error_check_ret is not None: |
| 217 | return error_check_ret |
| 218 | |
| 219 | if not request.generate_responses: |
| 220 | return self.create_error_response("generate_responses must not be empty") |
| 221 | |
| 222 | bounds_error = self._validate_derender_bounds(request.generate_responses) |
| 223 | if bounds_error is not None: |
| 224 | return bounds_error |
| 225 | |
| 226 | if request.prompt_token_ids is not None: |
| 227 | prompt_ids_error = self._validate_prompt_token_ids(request.prompt_token_ids) |
| 228 | if prompt_ids_error is not None: |
| 229 | return prompt_ids_error |
| 230 | |
| 231 | ( |
| 232 | choices, |
| 233 | total_prompt_tokens, |
| 234 | total_completion_tokens, |
| 235 | ) = await self.online_derenderer.derender_completion( |
| 236 | request.generate_responses, |
| 237 | request.prompt_tokens, |
| 238 | completion_request=request.completion_request, |
| 239 | prompt_token_ids=request.prompt_token_ids, |
| 240 | ) |
| 241 | |
| 242 | first = request.generate_responses[0] |
| 243 | kv_params = first.kv_transfer_params |
| 244 | if any( |
| 245 | r.kv_transfer_params != kv_params for r in request.generate_responses[1:] |
| 246 | ): |
| 247 | logger.warning( |
| 248 | "derender_completion: kv_transfer_params differ across responses; " |
| 249 | "setting to None on the aggregated response" |
| 250 | ) |
| 251 | kv_params = None |
| 252 | |
| 253 | usage = UsageInfo( |
| 254 | prompt_tokens=total_prompt_tokens, |
| 255 | completion_tokens=total_completion_tokens, |
| 256 | total_tokens=total_prompt_tokens + total_completion_tokens, |
| 257 | ) |
| 258 | |
| 259 | model_name = request.model or self.models.model_name() |
| 260 | logger.debug( |
| 261 | "derender_completion request_id=%s model=%s choices=%d" |
| 262 | " completion_tokens=%d", |
| 263 | first.request_id, |
| 264 | model_name, |
| 265 | len(choices), |
| 266 | total_completion_tokens, |
| 267 | ) |
| 268 | return CompletionResponse( |
| 269 | id=first.request_id, |
| 270 | model=model_name, |
| 271 | created=int(time.time()), |
| 272 | choices=choices, |
| 273 | usage=usage, |
| 274 | kv_transfer_params=kv_params, |
| 275 | # Metrics describe one prompt. Parallel samples are already |
| 276 | # suppressed by /generate; multi-prompt responses cannot be merged. |
| 277 | metrics=first.metrics if len(request.generate_responses) == 1 else None, |
| 278 | ) |
| 279 | |
| 280 | async def derender_chat_stream_response( |
| 281 | self, |
| 282 | request: DerenderChatStreamRequest, |
| 283 | ) -> tuple[ChatCompletionStreamResponse, DerenderStreamState] | ErrorResponse: |
| 284 | """Streaming counterpart to ``derender_chat_response``. |
| 285 | |
| 286 | Processes one ``GenerateStreamResponse`` chunk and returns the |
| 287 | derendered chunk together with the updated client carried state. |
| 288 | """ |
| 289 | error_check_ret = await self._check_model(request) |
| 290 | if error_check_ret is not None: |
| 291 | return error_check_ret |
| 292 | |
| 293 | if self.online_derenderer.parser is not None and request.chat_request is None: |
| 294 | return self.create_error_response( |
| 295 | "chat_request is required when a tool or reasoning parser is " |
| 296 | "configured because plain detokenization would leak raw parser " |
| 297 | "markup into content." |
| 298 | ) |
| 299 | |
| 300 | if ( |
| 301 | self.online_derenderer.parser is not None |
| 302 | and request.prompt_token_ids is None |
| 303 | ): |
| 304 | return self.create_error_response( |
| 305 | "prompt_token_ids is required when a tool or reasoning " |
| 306 | "parser is configured. Without it parse_delta cannot tell " |
| 307 | "whether the prompt left reasoning open (e.g. a chat " |
| 308 | "template that pre-opens <think>) and would misclassify " |
| 309 | "reasoning content as plain content." |
| 310 | ) |
| 311 | |
| 312 | # Each streamed chunk contains at most one choice per SSE event. |
| 313 | # A single DerenderStreamState is threaded |
| 314 | # through every choice in the chunk, so >1 would corrupt detok/parser |
| 315 | # state across choices. See the matching check in |
| 316 | # OnlineDerenderer.derender_chat_stream which this fails ahead of to |
| 317 | # avoid touching the tokenizer at all on a malformed chunk. |
| 318 | num_choices = len(request.generate_chunk.choices) |
| 319 | if num_choices > 1: |
| 320 | return self.create_error_response( |
| 321 | f"derender_chat_stream expects at most one choice per chunk " |
| 322 | f"(got {num_choices})." |
| 323 | ) |
| 324 | |
| 325 | # prompt_token_ids is folded in here too. It is caller supplied and |
| 326 | # otherwise unbounded. A parser configured deployment rescans it |
| 327 | # in full on every chunk (is_reasoning_end / |
| 328 | # adjust_initial_state_from_prompt, see abstract_parser.parse_delta) |
| 329 | # since parser state cannot be carried across calls. |
| 330 | prompt_len = ( |
| 331 | len(request.prompt_token_ids) if request.prompt_token_ids is not None else 0 |
| 332 | ) |
| 333 | prior_len = ( |
| 334 | len(request.stream_state.output_token_ids) |
| 335 | if request.stream_state is not None |
| 336 | else 0 |
| 337 | ) |
| 338 | delta_len = sum( |
| 339 | len(c.token_ids) for c in request.generate_chunk.choices if c.token_ids |
| 340 | ) |
| 341 | max_model_len = self.model_config.max_model_len |
| 342 | if prompt_len + prior_len + delta_len > max_model_len: |
| 343 | return self.create_error_response( |
| 344 | f"prompt_token_ids length ({prompt_len}) plus " |
| 345 | f"output_token_ids length ({prior_len}) plus delta " |
| 346 | f"({delta_len}) exceeds max_model_len ({max_model_len})." |
| 347 | ) |
| 348 | |
| 349 | model_name = request.model or self.models.model_name() |
| 350 | try: |
| 351 | chunk, updated_state = await self.online_derenderer.derender_chat_stream( |
| 352 | model=model_name, |
| 353 | generate_chunk=request.generate_chunk, |
| 354 | state=request.stream_state, |
| 355 | chat_request=request.chat_request, |
| 356 | prompt_tokens=request.prompt_tokens, |
| 357 | prompt_token_ids=request.prompt_token_ids, |
| 358 | ) |
| 359 | except Exception as exc: |
| 360 | # The two ValueErrors derender_chat_stream can raise directly |
| 361 | # (missing chat_request, >1 choice per chunk) are already |
| 362 | # pre-checked above, so anything reaching here comes from the |
| 363 | # parser replaying entirely client controlled state. Hermes |
| 364 | # swallows its own streaming exceptions but finalize_generation |
| 365 | # and engine based parsers don't, so an unexpected parser |
| 366 | # failure on malformed (but well typed) state must surface as |
| 367 | # a 400 here rather than an unhandled 500. |
| 368 | logger.debug( |
| 369 | "derender stream replay failed: %s: %s", type(exc).__name__, exc |
| 370 | ) |
| 371 | return self.create_error_response("invalid stream_state: derender failed") |
| 372 | |
| 373 | logger.debug( |
| 374 | "derender_chat_stream request_id=%s model=%s delta_tokens=%d", |
| 375 | request.generate_chunk.request_id, |
| 376 | model_name, |
| 377 | sum( |
| 378 | len(c.token_ids) for c in request.generate_chunk.choices if c.token_ids |
| 379 | ), |
| 380 | ) |
| 381 | return chunk, updated_state |
| 382 | |
| 383 | async def derender_completion_stream_response( |
| 384 | self, |
| 385 | request: DerenderCompletionStreamRequest, |
| 386 | ) -> tuple[CompletionStreamResponse, DerenderStreamState] | ErrorResponse: |
| 387 | """Streaming counterpart to ``derender_completion_response``. |
| 388 | |
| 389 | Processes one ``GenerateStreamResponse`` chunk (one output sequence's |
| 390 | delta) and returns the derendered chunk and updated state. |
| 391 | """ |
| 392 | error_check_ret = await self._check_model(request) |
| 393 | if error_check_ret is not None: |
| 394 | return error_check_ret |
| 395 | |
| 396 | prompt_ids_error = self._validate_prompt_token_ids([request.prompt_token_ids]) |
| 397 | if prompt_ids_error is not None: |
| 398 | return prompt_ids_error |
| 399 | |
| 400 | model_name = request.model or self.models.model_name() |
| 401 | try: |
| 402 | ( |
| 403 | chunk, |
| 404 | updated_state, |
| 405 | ) = await self.online_derenderer.derender_completion_stream( |
| 406 | model=model_name, |
| 407 | generate_chunk=request.generate_chunk, |
| 408 | state=request.stream_state, |
| 409 | prompt_tokens=request.prompt_tokens, |
| 410 | completion_request=request.completion_request, |
| 411 | prompt_token_ids=request.prompt_token_ids, |
| 412 | ) |
| 413 | except ValueError as exc: |
| 414 | return self.create_error_response(str(exc)) |
| 415 | except (KeyError, IndexError) as exc: |
| 416 | return self.create_error_response( |
| 417 | f"invalid stream_state: detokenization failed ({exc})" |
| 418 | ) |
| 419 | |
| 420 | logger.debug( |
| 421 | "derender_completion_stream request_id=%s model=%s delta_tokens=%d", |
| 422 | request.generate_chunk.request_id, |
| 423 | model_name, |
| 424 | sum( |
| 425 | len(c.token_ids) for c in request.generate_chunk.choices if c.token_ids |
| 426 | ), |
| 427 | ) |
| 428 | return chunk, updated_state |
| 429 | |
| 430 | @staticmethod |
| 431 | def _extract_mm_features( |
| 432 | engine_input: EngineInput, |
| 433 | ) -> MultiModalFeatures | None: |
| 434 | return extract_mm_features(engine_input) |