| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | |
| 5 | import asyncio |
| 6 | import time |
| 7 | from collections.abc import AsyncGenerator |
| 8 | from collections.abc import Sequence as GenericSequence |
| 9 | |
| 10 | import msgspec |
| 11 | from fastapi import Request |
| 12 | |
| 13 | from vllm.engine.protocol import EngineClient |
| 14 | from vllm.entrypoints.chat_utils import AsyncMultiModalItemTracker |
| 15 | from vllm.entrypoints.generate.base.protocol import ( |
| 16 | PerRequestMetrics, |
| 17 | RequestResponseMetadata, |
| 18 | ) |
| 19 | from vllm.entrypoints.generate.base.serving import ( |
| 20 | GenerateBaseServing, |
| 21 | build_spec_decoding_metrics, |
| 22 | clamp_prompt_logprobs, |
| 23 | ) |
| 24 | from vllm.entrypoints.openai.chat_completion.protocol import ( |
| 25 | ChatCompletionLogProb, |
| 26 | ChatCompletionLogProbs, |
| 27 | ChatCompletionLogProbsContent, |
| 28 | ) |
| 29 | from vllm.entrypoints.openai.models.serving import OpenAIServingModels |
| 30 | from vllm.entrypoints.serve.engine.protocol import ( |
| 31 | ErrorResponse, |
| 32 | PromptTokenUsageInfo, |
| 33 | UsageInfo, |
| 34 | ) |
| 35 | from vllm.entrypoints.serve.utils.api_utils import get_max_tokens, should_include_usage |
| 36 | from vllm.entrypoints.serve.utils.request_logger import RequestLogger |
| 37 | from vllm.exceptions import GenerationError |
| 38 | from vllm.inputs import EngineInput, TokensPrompt, mm_input |
| 39 | from vllm.logger import init_logger |
| 40 | from vllm.logprobs import Logprob |
| 41 | from vllm.multimodal.inputs import ( |
| 42 | MultiModalKwargsItem, |
| 43 | MultiModalKwargsItems, |
| 44 | PlaceholderRange, |
| 45 | ) |
| 46 | from vllm.outputs import RequestOutput |
| 47 | from vllm.renderers.online_renderer import OnlineRenderer |
| 48 | from vllm.sampling_params import RequestOutputKind, SamplingParams |
| 49 | from vllm.utils.collection_utils import as_list |
| 50 | from vllm.utils.serial_utils import numpy2base64 |
| 51 | |
| 52 | from .mm_features import ( |
| 53 | mm_kwargs_from_features, |
| 54 | placeholder_ranges_from_engine_input, |
| 55 | ) |
| 56 | from .protocol import ( |
| 57 | GenerateRequest, |
| 58 | GenerateResponse, |
| 59 | GenerateResponseChoice, |
| 60 | GenerateResponseStreamChoice, |
| 61 | GenerateStreamResponse, |
| 62 | ) |
| 63 | |
| 64 | logger = init_logger(__name__) |
| 65 | |
| 66 | |
| 67 | class ServingTokens(GenerateBaseServing): |
| 68 | """Provides Tokens IN <> Tokens OUT functionality to vLLM API.""" |
| 69 | |
| 70 | def __init__( |
| 71 | self, |
| 72 | engine_client: EngineClient, |
| 73 | models: OpenAIServingModels, |
| 74 | online_renderer: OnlineRenderer, |
| 75 | *, |
| 76 | request_logger: RequestLogger | None, |
| 77 | force_no_detokenize: bool = False, |
| 78 | return_tokens_as_token_ids: bool = False, |
| 79 | enable_prompt_tokens_details: bool = False, |
| 80 | enable_log_outputs: bool = False, |
| 81 | ): |
| 82 | super().__init__( |
| 83 | engine_client=engine_client, |
| 84 | models=models, |
| 85 | request_logger=request_logger, |
| 86 | return_tokens_as_token_ids=return_tokens_as_token_ids, |
| 87 | ) |
| 88 | self.online_renderer = online_renderer |
| 89 | self.enable_prompt_tokens_details = enable_prompt_tokens_details |
| 90 | self.enable_log_outputs = enable_log_outputs |
| 91 | self.force_no_detokenize = force_no_detokenize |
| 92 | if force_no_detokenize: |
| 93 | logger.info( |
| 94 | "Tokens-only mode is enabled, skipping detokenization " |
| 95 | "step for incoming requests." |
| 96 | ) |
| 97 | |
| 98 | # Mirrors ``OpenAIServingChat`` so we can apply server-side |
| 99 | # ``max_tokens`` defaulting when the client omits it. Without this, |
| 100 | # ``SamplingParams.max_tokens`` falls back to its dataclass default |
| 101 | # of 16 and silently truncates every generation. |
| 102 | self.default_sampling_params = self.model_config.get_diff_sampling_param() |
| 103 | mc = self.model_config |
| 104 | self.override_max_tokens = ( |
| 105 | self.default_sampling_params.get("max_tokens") |
| 106 | if mc.generation_config not in ("auto", "vllm") |
| 107 | else getattr(mc, "override_generation_config", {}).get("max_new_tokens") |
| 108 | ) |
| 109 | |
| 110 | def _validate_mm_cache_handles( |
| 111 | self, |
| 112 | mm_kwargs: dict[str, list[MultiModalKwargsItem | None]], |
| 113 | mm_hashes: dict[str, list[str]], |
| 114 | ) -> ErrorResponse | None: |
| 115 | cache = self.online_renderer.renderer.mm_processor_cache |
| 116 | if cache is None: |
| 117 | return None |
| 118 | try: |
| 119 | for modality, items in mm_kwargs.items(): |
| 120 | for mm_hash, item in zip(mm_hashes[modality], items, strict=True): |
| 121 | if item is not None: |
| 122 | cache.validate_input_item(item, mm_hash) |
| 123 | except ValueError as error: |
| 124 | return self.create_error_response(error) |
| 125 | return None |
| 126 | |
| 127 | async def serve_tokens( |
| 128 | self, |
| 129 | request: GenerateRequest, |
| 130 | raw_request: Request | None = None, |
| 131 | ) -> GenerateResponse | ErrorResponse | AsyncGenerator[str, None]: |
| 132 | error_check_ret = await self._check_model(request) |
| 133 | if error_check_ret is not None: |
| 134 | logger.error("Error with model %s", error_check_ret) |
| 135 | return error_check_ret |
| 136 | |
| 137 | self._preflight() |
| 138 | |
| 139 | lora_request = None |
| 140 | lora_request = self._maybe_get_adapters(request, supports_default_mm_loras=True) |
| 141 | |
| 142 | model_name = self.models.model_name(lora_request) |
| 143 | |
| 144 | request_id = ( |
| 145 | f"generate-tokens-{self._base_request_id(raw_request, request.request_id)}" |
| 146 | ) |
| 147 | |
| 148 | request_metadata = RequestResponseMetadata(request_id=request_id) |
| 149 | if raw_request: |
| 150 | raw_request.state.request_metadata = request_metadata |
| 151 | |
| 152 | sampling_params = request.sampling_params |
| 153 | max_num_seqs = self.engine_client.vllm_config.scheduler_config.max_num_seqs |
| 154 | if sampling_params.n > max_num_seqs: |
| 155 | return self.create_error_response( |
| 156 | f"sampling_params.n must be at most the server's max_num_seqs " |
| 157 | f"({max_num_seqs}), got {sampling_params.n}." |
| 158 | ) |
| 159 | # The stream schema has no field for the scores. |
| 160 | if request.stream and sampling_params.prompt_logprob_token_ids is not None: |
| 161 | return self.create_error_response( |
| 162 | "prompt_logprob_token_ids are not available when stream=true." |
| 163 | ) |
| 164 | if self.force_no_detokenize and sampling_params.stop: |
| 165 | # SamplingParams rejects stop with detokenize=False at request |
| 166 | # validation, but this server forces detokenize=False afterwards, |
| 167 | # so the combination must be rejected here or stop strings are |
| 168 | # silently never applied. |
| 169 | return self.create_error_response( |
| 170 | "stop strings are not supported on a --tokens-only server " |
| 171 | "because detokenization is disabled. Check stop strings on " |
| 172 | "the coordinator, or use stop_token_ids." |
| 173 | ) |
| 174 | try: |
| 175 | msgspec.msgpack.encode( |
| 176 | ( |
| 177 | sampling_params, |
| 178 | request.kv_transfer_params, |
| 179 | request.ec_transfer_params, |
| 180 | ) |
| 181 | ) |
| 182 | except (OverflowError, TypeError, ValueError) as e: |
| 183 | return self.create_error_response(e) |
| 184 | |
| 185 | engine_input: EngineInput |
| 186 | if request.content_parts: |
| 187 | tracker = AsyncMultiModalItemTracker(self.model_config) |
| 188 | mm_parser = tracker.create_parser() |
| 189 | for part in request.content_parts: |
| 190 | ptype = part.get("type", "") |
| 191 | url = part.get("url") |
| 192 | uuid = part.get("uuid") |
| 193 | if ptype == "image_url": |
| 194 | mm_parser.parse_image(url, uuid) |
| 195 | elif ptype == "audio_url": |
| 196 | mm_parser.parse_audio(url, uuid) |
| 197 | elif ptype == "video_url": |
| 198 | mm_parser.parse_video(url, uuid) |
| 199 | mm_data, mm_uuids = await tracker.resolve_items() |
| 200 | prompt = TokensPrompt(prompt_token_ids=request.token_ids) |
| 201 | if request.cache_salt is not None: |
| 202 | prompt["cache_salt"] = request.cache_salt |
| 203 | if mm_data: |
| 204 | prompt["multi_modal_data"] = mm_data |
| 205 | if mm_uuids: |
| 206 | prompt["multi_modal_uuids"] = mm_uuids |
| 207 | (engine_input,) = await self.online_renderer.renderer.render_cmpl_async( |
| 208 | [prompt] |
| 209 | ) |
| 210 | elif features := request.features: |
| 211 | # Convert PlaceholderRangeInfo → PlaceholderRange per modality. |
| 212 | mm_placeholders: dict[str, list[PlaceholderRange]] = { |
| 213 | modality: [ |
| 214 | PlaceholderRange(offset=p.offset, length=p.length) for p in ranges |
| 215 | ] |
| 216 | for modality, ranges in features.mm_placeholders.items() |
| 217 | } |
| 218 | |
| 219 | # Deserialize full tensor data and optional metadata-only data. |
| 220 | # Metadata-only items are valid when ec_transfer_params is set. |
| 221 | mm_kwargs = mm_kwargs_from_features(features) |
| 222 | if error := self._validate_mm_cache_handles(mm_kwargs, features.mm_hashes): |
| 223 | return error |
| 224 | |
| 225 | engine_input = mm_input( |
| 226 | prompt_token_ids=request.token_ids, |
| 227 | mm_kwargs=MultiModalKwargsItems(mm_kwargs), |
| 228 | mm_hashes=features.mm_hashes, |
| 229 | mm_placeholders=mm_placeholders, |
| 230 | cache_salt=request.cache_salt, |
| 231 | ) |
| 232 | else: |
| 233 | (engine_input,) = await self.online_renderer.preprocess_completion( |
| 234 | request, |
| 235 | prompt_input=request.token_ids, |
| 236 | prompt_embeds=None, |
| 237 | skip_mm_cache=True, |
| 238 | ) |
| 239 | |
| 240 | # Offsets are relative to the decoder prompt, so they are not |
| 241 | # meaningful for encoder-decoder models. |
| 242 | request._response_mm_placeholders = ( |
| 243 | placeholder_ranges_from_engine_input(engine_input) |
| 244 | if request.return_token_ids and not self.model_config.is_encoder_decoder |
| 245 | else None |
| 246 | ) |
| 247 | |
| 248 | # Schedule the request and get the result generator. |
| 249 | result_generator: AsyncGenerator[RequestOutput, None] | None = None |
| 250 | |
| 251 | # Pass disaggregated-serving parameters through to the engine. |
| 252 | if request.kv_transfer_params is not None: |
| 253 | extra = sampling_params.extra_args or {} |
| 254 | extra["kv_transfer_params"] = request.kv_transfer_params |
| 255 | sampling_params.extra_args = extra |
| 256 | if request.ec_transfer_params is not None: |
| 257 | extra = sampling_params.extra_args or {} |
| 258 | extra["ec_transfer_params"] = request.ec_transfer_params |
| 259 | sampling_params.extra_args = extra |
| 260 | |
| 261 | # Apply server-side ``max_tokens`` defaulting when the client did |
| 262 | # not set it, matching the OpenAI-compat endpoints. ``SamplingParams`` |
| 263 | # defaults ``max_tokens`` to 16, which would otherwise silently cap |
| 264 | # every generation that omits the field. |
| 265 | if not request.is_sampling_param_provided("max_tokens"): |
| 266 | sampling_params.max_tokens = get_max_tokens( |
| 267 | max_model_len=self.model_config.max_model_len, |
| 268 | max_tokens=None, |
| 269 | input_length=self._extract_prompt_len(engine_input), |
| 270 | default_sampling_params=self.default_sampling_params, |
| 271 | override_max_tokens=self.override_max_tokens, |
| 272 | ) |
| 273 | |
| 274 | if self.force_no_detokenize: |
| 275 | sampling_params.detokenize = False |
| 276 | sampling_params.output_kind = ( |
| 277 | RequestOutputKind.DELTA if request.stream else RequestOutputKind.FINAL_ONLY |
| 278 | ) |
| 279 | |
| 280 | self._log_inputs( |
| 281 | request_id, |
| 282 | engine_input, |
| 283 | params=sampling_params, |
| 284 | lora_request=lora_request, |
| 285 | ) |
| 286 | |
| 287 | trace_headers = ( |
| 288 | None |
| 289 | if raw_request is None |
| 290 | else await self._get_trace_headers(raw_request.headers) |
| 291 | ) |
| 292 | |
| 293 | # Extract data_parallel_rank from header (router can inject it) |
| 294 | data_parallel_rank = self._get_data_parallel_rank(raw_request) |
| 295 | session_id = self._get_session_id_from_headers(raw_request) |
| 296 | |
| 297 | result_generator = self.engine_client.generate( |
| 298 | engine_input, |
| 299 | sampling_params, |
| 300 | request_id, |
| 301 | lora_request=lora_request, |
| 302 | trace_headers=trace_headers, |
| 303 | priority=request.priority, |
| 304 | data_parallel_rank=data_parallel_rank, |
| 305 | session_id=session_id, |
| 306 | ) |
| 307 | |
| 308 | assert result_generator is not None |
| 309 | |
| 310 | if request.stream: |
| 311 | return self.serve_tokens_stream_generator( |
| 312 | request, |
| 313 | result_generator, |
| 314 | request_id, |
| 315 | model_name, |
| 316 | request_metadata, |
| 317 | ) |
| 318 | |
| 319 | return await self.serve_tokens_full_generator( |
| 320 | request, result_generator, request_id, model_name, request_metadata |
| 321 | ) |
| 322 | |
| 323 | async def serve_tokens_full_generator( |
| 324 | self, |
| 325 | request: GenerateRequest, |
| 326 | result_generator: AsyncGenerator[RequestOutput, None], |
| 327 | request_id: str, |
| 328 | model_name: str, |
| 329 | request_metadata: RequestResponseMetadata, |
| 330 | ) -> ErrorResponse | GenerateResponse: |
| 331 | created_time = int(time.time()) |
| 332 | final_res: RequestOutput | None = None |
| 333 | sampling_params: SamplingParams = request.sampling_params |
| 334 | |
| 335 | try: |
| 336 | async for res in result_generator: |
| 337 | final_res = res |
| 338 | except asyncio.CancelledError: |
| 339 | return self.create_error_response("Client disconnected") |
| 340 | |
| 341 | assert final_res is not None |
| 342 | |
| 343 | choices: list[GenerateResponseChoice] = [] |
| 344 | num_generated_tokens = 0 |
| 345 | for output in final_res.outputs: |
| 346 | self._raise_if_error(output.finish_reason, request_id) |
| 347 | |
| 348 | token_ids = output.token_ids |
| 349 | out_logprobs = output.logprobs |
| 350 | |
| 351 | # This is top_logprobs in completions API |
| 352 | if sampling_params.logprobs is not None: |
| 353 | assert out_logprobs is not None, "Did not output logprobs" |
| 354 | logprobs = self._create_tokens_logprobs( |
| 355 | token_ids=token_ids, |
| 356 | top_logprobs=out_logprobs, |
| 357 | num_output_top_logprobs=sampling_params.logprobs, |
| 358 | ) |
| 359 | else: |
| 360 | logprobs = None |
| 361 | |
| 362 | routed_experts_b64 = ( |
| 363 | numpy2base64(output.routed_experts) |
| 364 | if output.routed_experts is not None |
| 365 | else None |
| 366 | ) |
| 367 | |
| 368 | sampling_mask = None |
| 369 | if output.sampling_mask is not None: |
| 370 | sampling_mask = output.sampling_mask.token_ids |
| 371 | |
| 372 | choice_data = GenerateResponseChoice( |
| 373 | index=output.index, |
| 374 | logprobs=logprobs, |
| 375 | finish_reason=output.finish_reason if output.finish_reason else "stop", |
| 376 | token_ids=as_list(output.token_ids), |
| 377 | routed_experts=routed_experts_b64, |
| 378 | sampling_mask=sampling_mask, |
| 379 | ) |
| 380 | |
| 381 | choices.append(choice_data) |
| 382 | num_generated_tokens += len(output.token_ids) |
| 383 | |
| 384 | assert final_res.prompt_token_ids is not None |
| 385 | num_prompt_tokens = len(final_res.prompt_token_ids) |
| 386 | if final_res.encoder_prompt_token_ids is not None: |
| 387 | num_prompt_tokens += len(final_res.encoder_prompt_token_ids) |
| 388 | |
| 389 | usage = UsageInfo( |
| 390 | prompt_tokens=num_prompt_tokens, |
| 391 | completion_tokens=num_generated_tokens, |
| 392 | total_tokens=num_prompt_tokens + num_generated_tokens, |
| 393 | ) |
| 394 | if ( |
| 395 | self.enable_prompt_tokens_details |
| 396 | and final_res.num_cached_tokens is not None |
| 397 | ): |
| 398 | # This info is not available at the /coordinator level |
| 399 | usage.prompt_tokens_details = PromptTokenUsageInfo( |
| 400 | cached_tokens=final_res.num_cached_tokens |
| 401 | ) |
| 402 | |
| 403 | request_metadata.final_usage_info = usage |
| 404 | |
| 405 | per_request_metrics = None |
| 406 | if request.sampling_params.n == 1: |
| 407 | spec_stats = build_spec_decoding_metrics(final_res) |
| 408 | if spec_stats is not None: |
| 409 | per_request_metrics = PerRequestMetrics(speculative_decoding=spec_stats) |
| 410 | response = GenerateResponse( |
| 411 | request_id=request_id, |
| 412 | created=created_time, |
| 413 | model=model_name, |
| 414 | choices=choices, |
| 415 | usage=usage, |
| 416 | prompt_logprobs=clamp_prompt_logprobs(final_res.prompt_logprobs), |
| 417 | prompt_token_id_logprobs=( |
| 418 | numpy2base64(final_res.prompt_token_id_logprobs) |
| 419 | if final_res.prompt_token_id_logprobs is not None |
| 420 | else None |
| 421 | ), |
| 422 | prompt_token_ids=( |
| 423 | final_res.prompt_token_ids if request.return_token_ids else None |
| 424 | ), |
| 425 | mm_placeholders=request._response_mm_placeholders, |
| 426 | metrics=per_request_metrics, |
| 427 | kv_transfer_params=final_res.kv_transfer_params, |
| 428 | ec_transfer_params=final_res.ec_transfer_params, |
| 429 | ) |
| 430 | |
| 431 | # Log complete response if output logging is enabled |
| 432 | if self.enable_log_outputs and self.request_logger: |
| 433 | for choice in choices: |
| 434 | # Get the corresponding output token IDs |
| 435 | output_token_ids = None |
| 436 | if choice.index < len(final_res.outputs): |
| 437 | output_token_ids = final_res.outputs[choice.index].token_ids |
| 438 | |
| 439 | if output_token_ids: |
| 440 | # Log token_ids only. |
| 441 | self.request_logger.log_outputs( |
| 442 | request_id=request_id, |
| 443 | outputs="", |
| 444 | output_token_ids=output_token_ids, |
| 445 | finish_reason=choice.finish_reason, |
| 446 | is_streaming=False, |
| 447 | delta=False, |
| 448 | ) |
| 449 | |
| 450 | return response |
| 451 | |
| 452 | async def serve_tokens_stream_generator( |
| 453 | self, |
| 454 | request: GenerateRequest, |
| 455 | result_generator: AsyncGenerator[RequestOutput, None], |
| 456 | request_id: str, |
| 457 | model_name: str, |
| 458 | request_metadata: RequestResponseMetadata, |
| 459 | ) -> AsyncGenerator[str, None]: |
| 460 | num_prompt_tokens = 0 |
| 461 | num_generated_tokens: list[int] = [] |
| 462 | first_iteration = True |
| 463 | prompt_token_ids: list[int] | None = None |
| 464 | num_cached_tokens = None |
| 465 | sampling_params: SamplingParams = request.sampling_params |
| 466 | last_res: RequestOutput | None = None |
| 467 | |
| 468 | include_usage, include_continuous_usage = should_include_usage( |
| 469 | request.stream_options, False |
| 470 | ) |
| 471 | |
| 472 | try: |
| 473 | async for res in result_generator: |
| 474 | last_res = res |
| 475 | if first_iteration: |
| 476 | if res.prompt_token_ids is not None: |
| 477 | num_prompt_tokens = len(res.prompt_token_ids) |
| 478 | if request.return_token_ids: |
| 479 | prompt_token_ids = res.prompt_token_ids |
| 480 | if res.encoder_prompt_token_ids is not None: |
| 481 | num_prompt_tokens += len(res.encoder_prompt_token_ids) |
| 482 | num_cached_tokens = res.num_cached_tokens |
| 483 | num_generated_tokens = [0] * len(res.outputs) |
| 484 | first_iteration = False |
| 485 | |
| 486 | for output in res.outputs: |
| 487 | i = output.index |
| 488 | delta_token_ids = output.token_ids |
| 489 | num_generated_tokens[i] += len(delta_token_ids) |
| 490 | |
| 491 | finish_reason = output.finish_reason |
| 492 | self._raise_if_error(finish_reason, request_id) |
| 493 | |
| 494 | # Still emit a terminal empty chunk while prompt metadata |
| 495 | # is pending, so zero-token completions deliver it. |
| 496 | if not delta_token_ids and ( |
| 497 | finish_reason is None or prompt_token_ids is None |
| 498 | ): |
| 499 | continue |
| 500 | |
| 501 | if sampling_params.logprobs is not None: |
| 502 | out_logprobs = output.logprobs |
| 503 | assert out_logprobs is not None, "Did not output logprobs" |
| 504 | logprobs = self._create_tokens_logprobs( |
| 505 | token_ids=delta_token_ids, |
| 506 | top_logprobs=out_logprobs, |
| 507 | num_output_top_logprobs=sampling_params.logprobs, |
| 508 | ) |
| 509 | else: |
| 510 | logprobs = None |
| 511 | |
| 512 | routed_experts_b64 = ( |
| 513 | numpy2base64(output.routed_experts) |
| 514 | if output.routed_experts is not None |
| 515 | else None |
| 516 | ) |
| 517 | |
| 518 | sampling_mask = None |
| 519 | if output.sampling_mask is not None: |
| 520 | sampling_mask = output.sampling_mask.token_ids |
| 521 | |
| 522 | chunk = GenerateStreamResponse( |
| 523 | request_id=request_id, |
| 524 | choices=[ |
| 525 | GenerateResponseStreamChoice( |
| 526 | index=i, |
| 527 | logprobs=logprobs, |
| 528 | finish_reason=finish_reason, |
| 529 | token_ids=as_list(delta_token_ids), |
| 530 | routed_experts=routed_experts_b64, |
| 531 | sampling_mask=sampling_mask, |
| 532 | ) |
| 533 | ], |
| 534 | ) |
| 535 | if prompt_token_ids is not None: |
| 536 | chunk.prompt_token_ids = prompt_token_ids |
| 537 | chunk.mm_placeholders = request._response_mm_placeholders |
| 538 | prompt_token_ids = None |
| 539 | if include_continuous_usage: |
| 540 | chunk.usage = UsageInfo( |
| 541 | prompt_tokens=num_prompt_tokens, |
| 542 | completion_tokens=num_generated_tokens[i], |
| 543 | total_tokens=(num_prompt_tokens + num_generated_tokens[i]), |
| 544 | ) |
| 545 | |
| 546 | # Omit fields that are absent from token-bearing chunks. |
| 547 | exclude = { |
| 548 | name |
| 549 | for name in ("prompt_token_ids", "mm_placeholders", "metrics") |
| 550 | if getattr(chunk, name) is None |
| 551 | } |
| 552 | yield f"data: {chunk.model_dump_json(exclude=exclude)}\n\n" |
| 553 | |
| 554 | total_completion_tokens = sum(num_generated_tokens) |
| 555 | final_usage_info = UsageInfo( |
| 556 | prompt_tokens=num_prompt_tokens, |
| 557 | completion_tokens=total_completion_tokens, |
| 558 | total_tokens=num_prompt_tokens + total_completion_tokens, |
| 559 | ) |
| 560 | |
| 561 | if self.enable_prompt_tokens_details and num_cached_tokens is not None: |
| 562 | final_usage_info.prompt_tokens_details = PromptTokenUsageInfo( |
| 563 | cached_tokens=num_cached_tokens |
| 564 | ) |
| 565 | |
| 566 | if include_usage: |
| 567 | per_request_metrics = None |
| 568 | if sampling_params.n == 1: |
| 569 | spec_stats = build_spec_decoding_metrics(last_res) |
| 570 | if spec_stats is not None: |
| 571 | per_request_metrics = PerRequestMetrics( |
| 572 | speculative_decoding=spec_stats |
| 573 | ) |
| 574 | final_chunk = GenerateStreamResponse( |
| 575 | request_id=request_id, |
| 576 | choices=[], |
| 577 | usage=final_usage_info, |
| 578 | metrics=per_request_metrics, |
| 579 | ) |
| 580 | yield f"data: {final_chunk.model_dump_json(exclude_none=True)}\n\n" |
| 581 | |
| 582 | request_metadata.final_usage_info = final_usage_info |
| 583 | |
| 584 | except GenerationError as e: |
| 585 | yield ( |
| 586 | f"data: {self._convert_generation_error_to_streaming_response(e)}\n\n" |
| 587 | ) |
| 588 | except Exception as e: |
| 589 | logger.exception("Error in token generation stream.") |
| 590 | data = self.create_streaming_error_response(e) |
| 591 | yield f"data: {data}\n\n" |
| 592 | yield "data: [DONE]\n\n" |
| 593 | |
| 594 | def _create_tokens_logprobs( |
| 595 | self, |
| 596 | token_ids: GenericSequence[int], |
| 597 | top_logprobs: GenericSequence[dict[int, Logprob] | None], |
| 598 | num_output_top_logprobs: int | None = None, |
| 599 | ) -> ChatCompletionLogProbs: |
| 600 | """Create OpenAI-style logprobs.""" |
| 601 | logprobs_content: list[ChatCompletionLogProbsContent] = [] |
| 602 | |
| 603 | for i, token_id in enumerate(token_ids): |
| 604 | token = f"token_id:{token_id}" |
| 605 | step_top_logprobs = top_logprobs[i] |
| 606 | if step_top_logprobs is None or step_top_logprobs.get(token_id) is None: |
| 607 | logprobs_content.append( |
| 608 | ChatCompletionLogProbsContent( |
| 609 | token=token, |
| 610 | ) |
| 611 | ) |
| 612 | else: |
| 613 | step_token = step_top_logprobs[token_id] |
| 614 | |
| 615 | logprobs_content.append( |
| 616 | ChatCompletionLogProbsContent( |
| 617 | token=token, |
| 618 | logprob=max(step_token.logprob, -9999.0), |
| 619 | top_logprobs=[ |
| 620 | ChatCompletionLogProb( |
| 621 | token=f"token_id:{token_id}", |
| 622 | logprob=max(logprob.logprob, -9999.0), |
| 623 | ) |
| 624 | for i, (token_id, logprob) in enumerate( |
| 625 | step_top_logprobs.items() |
| 626 | ) |
| 627 | if num_output_top_logprobs is not None |
| 628 | and ( |
| 629 | num_output_top_logprobs == -1 |
| 630 | or i < max(num_output_top_logprobs, 1) |
| 631 | ) |
| 632 | ], |
| 633 | ) |
| 634 | ) |
| 635 | |
| 636 | return ChatCompletionLogProbs(content=logprobs_content) |