| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import math |
| 5 | import time |
| 6 | from unittest.mock import MagicMock, Mock |
| 7 | |
| 8 | import numpy as np |
| 9 | import pytest |
| 10 | import torch |
| 11 | |
| 12 | from tests.v1.engine.utils import ( |
| 13 | NUM_PROMPT_LOGPROBS_UNDER_TEST, |
| 14 | NUM_SAMPLE_LOGPROBS_UNDER_TEST, |
| 15 | STOP_STRINGS, |
| 16 | DummyOutputProcessorTestVectors, |
| 17 | MockEngineCore, |
| 18 | ) |
| 19 | from vllm import PoolingParams |
| 20 | from vllm.logprobs import FlatLogprobs, Logprob, PromptLogprobs, SampleLogprobs |
| 21 | from vllm.lora.request import LoRARequest |
| 22 | from vllm.outputs import ( |
| 23 | CompletionOutput, |
| 24 | PoolingRequestOutput, |
| 25 | RequestOutput, |
| 26 | SamplingMask, |
| 27 | ) |
| 28 | from vllm.sampling_params import RequestOutputKind, SamplingParams |
| 29 | from vllm.tokenizers import TokenizerLike |
| 30 | from vllm.v1.engine import ( |
| 31 | EngineCoreEvent, |
| 32 | EngineCoreEventType, |
| 33 | EngineCoreOutput, |
| 34 | EngineCoreOutputs, |
| 35 | EngineCoreRequest, |
| 36 | FinishReason, |
| 37 | ) |
| 38 | from vllm.v1.engine.output_processor import ( |
| 39 | OutputProcessor, |
| 40 | RequestOutputCollector, |
| 41 | RequestState, |
| 42 | ) |
| 43 | from vllm.v1.metrics.stats import IterationStats, PrefillStats, SchedulerStats |
| 44 | from vllm.v1.outputs import SamplingMaskLists |
| 45 | |
| 46 | |
| 47 | @pytest.mark.parametrize("flat_logprobs", [False, True]) |
| 48 | def test_delta_output_without_new_tokens_returns_empty_logprobs( |
| 49 | flat_logprobs: bool, |
| 50 | ) -> None: |
| 51 | accumulated = FlatLogprobs() if flat_logprobs else [] |
| 52 | accumulated.append({1: Logprob(logprob=-0.5, rank=1)}) |
| 53 | |
| 54 | state = RequestState.__new__(RequestState) |
| 55 | state.detokenizer = MagicMock() |
| 56 | state.detokenizer.get_next_output_text.return_value = "" |
| 57 | state.logprobs_processor = MagicMock() |
| 58 | state.logprobs_processor.logprobs = accumulated |
| 59 | state.logprobs_processor.cumulative_logprob = -0.5 |
| 60 | state.output_kind = RequestOutputKind.DELTA |
| 61 | state.request_index = 0 |
| 62 | state.sampling_mask_chunks = [] |
| 63 | state.routed_experts_chunks = [] |
| 64 | state.spec_decode_metrics = None |
| 65 | |
| 66 | output = state._new_completion_output([], None, None) |
| 67 | |
| 68 | assert isinstance(output.logprobs, FlatLogprobs if flat_logprobs else list) |
| 69 | assert len(output.logprobs) == 0 |
| 70 | |
| 71 | |
| 72 | def test_completion_output_preserves_each_sampling_mask_position() -> None: |
| 73 | state = RequestState.__new__(RequestState) |
| 74 | state.detokenizer = MagicMock() |
| 75 | state.detokenizer.get_next_output_text.return_value = "" |
| 76 | state.logprobs_processor = MagicMock() |
| 77 | state.logprobs_processor.logprobs = None |
| 78 | state.logprobs_processor.cumulative_logprob = None |
| 79 | state.output_kind = RequestOutputKind.DELTA |
| 80 | state.request_index = 0 |
| 81 | state.sampling_mask_chunks = [ |
| 82 | SamplingMaskLists( |
| 83 | token_ids=np.array([10, 11, 20]), |
| 84 | offsets=np.array([0, 2, 3]), |
| 85 | ), |
| 86 | SamplingMaskLists(token_ids=np.array([30, 31, 32])), |
| 87 | ] |
| 88 | state.routed_experts_chunks = [] |
| 89 | state.spec_decode_metrics = None |
| 90 | |
| 91 | output = state._new_completion_output([1, 2, 3], FinishReason.LENGTH, None) |
| 92 | |
| 93 | assert output.sampling_mask is not None |
| 94 | assert output.sampling_mask.token_ids == [[10, 11], [20], [30, 31, 32]] |
| 95 | |
| 96 | |
| 97 | def _ref_convert_id_to_token( |
| 98 | tokenizer: TokenizerLike, |
| 99 | token_id: int, |
| 100 | ) -> str: |
| 101 | """Reference impl of logprobs detokenization. |
| 102 | |
| 103 | Args: |
| 104 | tokenizer: tokenizer used by the model under test |
| 105 | token_id: convert this token id |
| 106 | |
| 107 | Returns: |
| 108 | String representation of input token id |
| 109 | |
| 110 | """ |
| 111 | return tokenizer.decode([token_id]) or "" |
| 112 | |
| 113 | |
| 114 | @pytest.mark.parametrize( |
| 115 | "request_output_kind", [RequestOutputKind.DELTA, RequestOutputKind.FINAL_ONLY] |
| 116 | ) |
| 117 | @pytest.mark.parametrize("stream_interval", [1, 5, 10]) |
| 118 | def test_incremental_detokenization( |
| 119 | request_output_kind: RequestOutputKind, |
| 120 | stream_interval: int, |
| 121 | dummy_test_vectors, |
| 122 | ): |
| 123 | output_processor = OutputProcessor( |
| 124 | dummy_test_vectors.tokenizer, log_stats=False, stream_interval=stream_interval |
| 125 | ) |
| 126 | |
| 127 | # Make N requests. |
| 128 | requests = [ |
| 129 | EngineCoreRequest( |
| 130 | request_id=f"request-{idx}-int", |
| 131 | external_req_id=f"request-{idx}", |
| 132 | prompt_token_ids=prompt_tokens, |
| 133 | mm_features=None, |
| 134 | arrival_time=0, |
| 135 | lora_request=None, |
| 136 | cache_salt=None, |
| 137 | data_parallel_rank=None, |
| 138 | sampling_params=SamplingParams( |
| 139 | skip_special_tokens=False, |
| 140 | spaces_between_special_tokens=False, |
| 141 | output_kind=request_output_kind, |
| 142 | stop=[], |
| 143 | include_stop_str_in_output=False, |
| 144 | ), |
| 145 | pooling_params=None, |
| 146 | ) |
| 147 | for idx, prompt_tokens in enumerate(dummy_test_vectors.prompt_tokens) |
| 148 | ] |
| 149 | |
| 150 | engine_core = MockEngineCore( |
| 151 | tokens_list=dummy_test_vectors.generation_tokens, |
| 152 | prompts_list=dummy_test_vectors.prompt_tokens, |
| 153 | request_ids=[req.request_id for req in requests], |
| 154 | ) |
| 155 | |
| 156 | # Add requests to the detokenizer. |
| 157 | for request, prompt in zip(requests, dummy_test_vectors.prompt_strings): |
| 158 | output_processor.add_request(request, prompt) |
| 159 | |
| 160 | gen_strings = {} |
| 161 | gen_tokens = {} |
| 162 | while True: |
| 163 | # Mock output from the EngineCore. |
| 164 | outputs = engine_core.get_outputs() |
| 165 | if len(outputs) == 0: |
| 166 | break |
| 167 | |
| 168 | # Step the Detokenizer. |
| 169 | processed_outputs = output_processor.process_outputs(outputs) |
| 170 | request_outputs = processed_outputs.request_outputs |
| 171 | requests_to_abort = processed_outputs.reqs_to_abort |
| 172 | assert len(requests_to_abort) == 0 |
| 173 | |
| 174 | # Update tracking. |
| 175 | for request_output in request_outputs: |
| 176 | request_id = request_output.request_id |
| 177 | new_text = request_output.outputs[0].text |
| 178 | new_tokens = request_output.outputs[0].token_ids |
| 179 | if request_id not in gen_strings: |
| 180 | gen_strings[request_id] = new_text |
| 181 | gen_tokens[request_id] = new_tokens |
| 182 | if request_output_kind == RequestOutputKind.DELTA: |
| 183 | assert len(new_tokens) == 1, f"{len(new_tokens)=}" |
| 184 | else: |
| 185 | gen_strings[request_id] += new_text |
| 186 | gen_tokens[request_id].extend(new_tokens) |
| 187 | if ( |
| 188 | request_output_kind == RequestOutputKind.DELTA |
| 189 | and not request_output.finished |
| 190 | ): |
| 191 | assert len(new_tokens) >= stream_interval, ( |
| 192 | f"{len(new_tokens)=}, {stream_interval=}" |
| 193 | ) |
| 194 | |
| 195 | # Confirmed tracked values matches what we expected. |
| 196 | for idx, (ref_gen_str, ref_gen_toks) in enumerate( |
| 197 | zip(dummy_test_vectors.generation_strings, dummy_test_vectors.generation_tokens) |
| 198 | ): |
| 199 | gen_str = gen_strings[f"request-{idx}"] |
| 200 | gen_toks = gen_tokens[f"request-{idx}"] |
| 201 | |
| 202 | assert gen_str == ref_gen_str, f"{gen_str=}, {ref_gen_str=}" |
| 203 | assert gen_toks == ref_gen_toks, f"{gen_toks=}, {ref_gen_toks=}" |
| 204 | |
| 205 | assert output_processor.get_num_unfinished_requests() == 0 |
| 206 | assert not output_processor.has_unfinished_requests() |
| 207 | |
| 208 | |
| 209 | @pytest.mark.parametrize("do_remote_prefill", [True, False]) |
| 210 | def test_remote_prefill_cached_tokens_override(do_remote_prefill: bool): |
| 211 | """P/D disaggregation: num_cached_tokens should report the P worker's |
| 212 | cache hits (passed via kv_transfer_params) instead of the local count, |
| 213 | which sees the KVs pulled from the remote prefill as a ~100% hit. |
| 214 | """ |
| 215 | output_processor = OutputProcessor(tokenizer=None, log_stats=False) |
| 216 | |
| 217 | prompt_tokens = [1, 2, 3, 4, 5, 6, 7, 8] |
| 218 | kv_transfer_params = { |
| 219 | "do_remote_prefill": do_remote_prefill, |
| 220 | "remote_prefill_cached_tokens": 5, |
| 221 | } |
| 222 | request = EngineCoreRequest( |
| 223 | request_id="request-0-int", |
| 224 | external_req_id="request-0", |
| 225 | prompt_token_ids=prompt_tokens, |
| 226 | mm_features=None, |
| 227 | arrival_time=0, |
| 228 | lora_request=None, |
| 229 | cache_salt=None, |
| 230 | data_parallel_rank=None, |
| 231 | sampling_params=SamplingParams( |
| 232 | detokenize=False, |
| 233 | extra_args={"kv_transfer_params": kv_transfer_params}, |
| 234 | ), |
| 235 | pooling_params=None, |
| 236 | ) |
| 237 | output_processor.add_request(request, prompt=None) |
| 238 | |
| 239 | prefill_stats = PrefillStats() |
| 240 | prefill_stats.set( |
| 241 | num_prompt_tokens=len(prompt_tokens), |
| 242 | num_local_cached_tokens=0, |
| 243 | num_external_cached_tokens=len(prompt_tokens) - 1, |
| 244 | ) |
| 245 | processed = output_processor.process_outputs( |
| 246 | [ |
| 247 | EngineCoreOutput( |
| 248 | request_id="request-0-int", |
| 249 | new_token_ids=[42], |
| 250 | prefill_stats=prefill_stats, |
| 251 | ) |
| 252 | ] |
| 253 | ) |
| 254 | request_output = processed.request_outputs[0] |
| 255 | if do_remote_prefill: |
| 256 | assert request_output.num_cached_tokens == 5 |
| 257 | else: |
| 258 | assert request_output.num_cached_tokens == len(prompt_tokens) - 1 |
| 259 | |
| 260 | |
| 261 | def test_request_stream_interval_raises_but_not_below_engine_default( |
| 262 | dummy_test_vectors, |
| 263 | ): |
| 264 | """A per-request stream_interval can raise the interval above the engine |
| 265 | default but not below it (values under the default clamp up), without |
| 266 | altering the generated text.""" |
| 267 | engine_stream_interval = 5 |
| 268 | # Request 0 (below the default) clamps up to 5; request 1 raises it to 10. |
| 269 | request_stream_intervals = [1, 10] |
| 270 | output_processor = OutputProcessor( |
| 271 | dummy_test_vectors.tokenizer, |
| 272 | log_stats=False, |
| 273 | stream_interval=engine_stream_interval, |
| 274 | ) |
| 275 | |
| 276 | requests = [ |
| 277 | EngineCoreRequest( |
| 278 | request_id=f"request-{idx}-int", |
| 279 | external_req_id=f"request-{idx}", |
| 280 | prompt_token_ids=prompt_tokens, |
| 281 | mm_features=None, |
| 282 | arrival_time=0, |
| 283 | lora_request=None, |
| 284 | cache_salt=None, |
| 285 | data_parallel_rank=None, |
| 286 | sampling_params=SamplingParams( |
| 287 | skip_special_tokens=False, |
| 288 | spaces_between_special_tokens=False, |
| 289 | output_kind=RequestOutputKind.DELTA, |
| 290 | stop=[], |
| 291 | include_stop_str_in_output=False, |
| 292 | stream_interval=request_stream_intervals[idx], |
| 293 | ), |
| 294 | pooling_params=None, |
| 295 | ) |
| 296 | for idx, prompt_tokens in enumerate( |
| 297 | dummy_test_vectors.prompt_tokens[: len(request_stream_intervals)] |
| 298 | ) |
| 299 | ] |
| 300 | |
| 301 | num_requests = len(requests) |
| 302 | engine_core = MockEngineCore( |
| 303 | tokens_list=dummy_test_vectors.generation_tokens[:num_requests], |
| 304 | prompts_list=dummy_test_vectors.prompt_tokens[:num_requests], |
| 305 | request_ids=[req.request_id for req in requests], |
| 306 | ) |
| 307 | |
| 308 | for request, prompt in zip(requests, dummy_test_vectors.prompt_strings): |
| 309 | output_processor.add_request(request, prompt) |
| 310 | |
| 311 | gen_strings: dict[str, str] = {} |
| 312 | gen_tokens: dict[str, list[int]] = {} |
| 313 | while outputs := engine_core.get_outputs(): |
| 314 | for request_output in output_processor.process_outputs(outputs).request_outputs: |
| 315 | request_id = request_output.request_id |
| 316 | new_tokens = request_output.outputs[0].token_ids |
| 317 | if request_id not in gen_strings: |
| 318 | gen_strings[request_id] = request_output.outputs[0].text |
| 319 | gen_tokens[request_id] = list(new_tokens) |
| 320 | assert len(new_tokens) == 1, f"{len(new_tokens)=}" |
| 321 | continue |
| 322 | gen_strings[request_id] += request_output.outputs[0].text |
| 323 | gen_tokens[request_id].extend(new_tokens) |
| 324 | if not request_output.finished: |
| 325 | requested = request_stream_intervals[int(request_id.split("-")[1])] |
| 326 | interval = max(requested, engine_stream_interval) |
| 327 | assert len(new_tokens) == interval, f"{len(new_tokens)=}, {interval=}" |
| 328 | |
| 329 | for idx in range(num_requests): |
| 330 | request_id = f"request-{idx}" |
| 331 | assert gen_strings[request_id] == dummy_test_vectors.generation_strings[idx] |
| 332 | assert gen_tokens[request_id] == dummy_test_vectors.generation_tokens[idx] |
| 333 | |
| 334 | assert not output_processor.has_unfinished_requests() |
| 335 | |
| 336 | |
| 337 | def _validate_logprobs( |
| 338 | gen_tokens: dict[str, list[int]], |
| 339 | gen_logprobs: dict[str, SampleLogprobs | None], |
| 340 | gen_prompt_logprobs: dict[str, PromptLogprobs | None], |
| 341 | gen_cumulative_logprob: dict[str, float], |
| 342 | dtv: DummyOutputProcessorTestVectors, |
| 343 | request_id_list: list[str], |
| 344 | num_sample_logprobs: int | None, |
| 345 | num_prompt_logprobs: int | None, |
| 346 | ) -> None: |
| 347 | for req_idx, req_id in enumerate(request_id_list): |
| 348 | new_tokens = gen_tokens[req_id] |
| 349 | logprobs = gen_logprobs[req_id] |
| 350 | prompt_logprobs = gen_prompt_logprobs[req_id] |
| 351 | cumulative_logprob = gen_cumulative_logprob[req_id] |
| 352 | prompt_token_ids = dtv.prompt_tokens[req_idx] |
| 353 | ref_logprobs = dtv.generation_logprobs[req_idx] |
| 354 | ref_prompt_logprobs = dtv.prompt_logprobs[req_idx] |
| 355 | if num_sample_logprobs is not None: |
| 356 | # Validate sample logprobs |
| 357 | assert logprobs is not None, ( |
| 358 | f"Request {req_id} requires sample" |
| 359 | " logprobs but sample logprobs are" |
| 360 | " None." |
| 361 | ) |
| 362 | # Require num sampled tokens to match num |
| 363 | # sampled logprobs - especially important |
| 364 | # to check since the detokenizer can cause |
| 365 | # a request to finish early due to a stop |
| 366 | # string being hit |
| 367 | num_new_tokens = len(new_tokens) |
| 368 | len_sample_logprobs = len(logprobs) |
| 369 | assert num_new_tokens == len_sample_logprobs, ( |
| 370 | f"Request {req_id} has {num_new_tokens}" |
| 371 | " completion tokens but has" |
| 372 | f" {len_sample_logprobs} sample logprobs." |
| 373 | ) |
| 374 | ref_cumulative_logprob = 0.0 |
| 375 | for idx, (sampled_token, pos_logprob_dict) in enumerate( |
| 376 | zip(new_tokens, logprobs) |
| 377 | ): |
| 378 | # Break out the reference log probability value & |
| 379 | # logprob token id tensors associated with this |
| 380 | # position in the completion. Also break out the |
| 381 | # sampled token ranks |
| 382 | (ref_pos_logprob_toks, ref_pos_logprob_vals, ref_sampled_token_rank) = ( |
| 383 | ref_logprobs[idx] |
| 384 | ) |
| 385 | # For each position in the completion sequence, |
| 386 | # ensure the actual sampled token is among the |
| 387 | # logprobs |
| 388 | assert sampled_token in pos_logprob_dict, ( |
| 389 | f"Sampled token {sampled_token} not" |
| 390 | f" present in logprob at index {idx}" |
| 391 | ) |
| 392 | |
| 393 | # Validate number of sample logprobs |
| 394 | num_lp_toks = len(pos_logprob_dict) |
| 395 | assert ( |
| 396 | num_lp_toks == num_sample_logprobs |
| 397 | or num_lp_toks == num_sample_logprobs + 1 |
| 398 | ), ( |
| 399 | "Valid numbers of sample logprobs are" |
| 400 | f" {num_sample_logprobs} or" |
| 401 | f" {num_sample_logprobs + 1} but" |
| 402 | f" {num_lp_toks} logprobs found at" |
| 403 | f" position {idx}. Logprobs dict:" |
| 404 | f" {pos_logprob_dict}" |
| 405 | ) |
| 406 | |
| 407 | # Validate sampled token logprob rank |
| 408 | smp_lp = pos_logprob_dict[sampled_token] |
| 409 | smp_lp_rank = smp_lp.rank |
| 410 | assert ref_sampled_token_rank == smp_lp_rank, ( |
| 411 | "Sampled token logprob rank" |
| 412 | f" {smp_lp_rank} does not match" |
| 413 | " correct value" |
| 414 | f" {ref_sampled_token_rank}" |
| 415 | f" in Logprob {smp_lp}" |
| 416 | ) |
| 417 | |
| 418 | # Validate that the logprob processor yields |
| 419 | # the correct log probabilities and valid |
| 420 | # rankings |
| 421 | rank_one_appears = False |
| 422 | for jdx in range(1, len(ref_pos_logprob_toks)): |
| 423 | # Iterate over the (logprob val,logprob tok id) |
| 424 | # pairs expected by the test fixture at this |
| 425 | # position in the completion. |
| 426 | ref_lp_val = ref_pos_logprob_vals[jdx] |
| 427 | ref_tok_id = ref_pos_logprob_toks[jdx] |
| 428 | assert ref_tok_id in pos_logprob_dict, ( |
| 429 | f"Expected token {ref_tok_id} to be" |
| 430 | f" in logprob dict but it is not." |
| 431 | ) |
| 432 | |
| 433 | # Extract actually-generated logprob |
| 434 | # info |
| 435 | lp = pos_logprob_dict[ref_tok_id] |
| 436 | lp_val = lp.logprob |
| 437 | lp_rank = lp.rank |
| 438 | |
| 439 | # A "top" (rank 1) logprob must be |
| 440 | # present |
| 441 | rank_one_appears = True if lp_rank == 1 else rank_one_appears |
| 442 | |
| 443 | # Rank must be >= 1 |
| 444 | assert lp_rank >= 1, ( |
| 445 | f"Logprob {lp} has invalid" |
| 446 | f" rank {lp_rank} < 1." |
| 447 | f" Logprob dict: {pos_logprob_dict}" |
| 448 | ) |
| 449 | |
| 450 | # Validate log probability |
| 451 | assert math.isclose(lp_val, ref_lp_val), ( |
| 452 | f"Token id {ref_tok_id} appears in logprobs dict" |
| 453 | f" at position {idx} in completion with log" |
| 454 | f" probability {lp_val} but {ref_lp_val} was" |
| 455 | f" expected. Logprob: {lp}" |
| 456 | ) |
| 457 | |
| 458 | assert rank_one_appears, ( |
| 459 | f"No Logprob has rank 1" |
| 460 | " in the following Logprob" |
| 461 | f" dict: {pos_logprob_dict}" |
| 462 | ) |
| 463 | |
| 464 | # Validate logprobs detokenization |
| 465 | for lp_tok in pos_logprob_dict: |
| 466 | # Confirm that sample logprob decoded token matches |
| 467 | # the logprob token id at this sequence position |
| 468 | decoded_token = pos_logprob_dict[lp_tok].decoded_token |
| 469 | ref_decoded_token = _ref_convert_id_to_token(dtv.tokenizer, lp_tok) |
| 470 | |
| 471 | # With UTF-8 correction logic, tokens ending with "�" |
| 472 | # (incomplete byte sequences) are corrected to either |
| 473 | # empty string or proper UTF-8 characters |
| 474 | if ref_decoded_token.endswith("�"): |
| 475 | # Token needs UTF-8 correction |
| 476 | assert not decoded_token.endswith("�"), ( |
| 477 | f"Sampled logprob token id {lp_tok} decodes to" |
| 478 | f" '{ref_decoded_token}' (ends with replacement char)" |
| 479 | f" but corrected decoded token '{decoded_token}'" |
| 480 | f" still ends with replacement char" |
| 481 | f" (at position {idx}). UTF-8 correction should" |
| 482 | f" have removed it." |
| 483 | ) |
| 484 | else: |
| 485 | # No correction needed, should match exactly |
| 486 | assert decoded_token == ref_decoded_token, ( |
| 487 | f"Sampled logprob token id {lp_tok} decodes to" |
| 488 | f" {ref_decoded_token} but Logprob decoded" |
| 489 | f" token is {decoded_token} instead" |
| 490 | f" (at position {idx})" |
| 491 | ) |
| 492 | |
| 493 | ref_cumulative_logprob += pos_logprob_dict[sampled_token].logprob |
| 494 | # Assert that cumulative logprobs are correct |
| 495 | assert math.isclose(cumulative_logprob, ref_cumulative_logprob) |
| 496 | else: |
| 497 | # Sample logprobs disabled for this request |
| 498 | assert logprobs is None |
| 499 | assert cumulative_logprob is None |
| 500 | |
| 501 | if num_prompt_logprobs is not None: |
| 502 | # Validate prompt logprobs |
| 503 | assert prompt_logprobs is not None, ( |
| 504 | f"Request {req_id} requires prompt" |
| 505 | " logprobs but prompt logprobs are" |
| 506 | " None." |
| 507 | ) |
| 508 | # Require num prompt tokens to match num |
| 509 | # prompt logprobs |
| 510 | num_prompt_tokens = len(prompt_token_ids) |
| 511 | len_prompt_logprobs = len(prompt_logprobs) |
| 512 | assert num_prompt_tokens == len_prompt_logprobs, ( |
| 513 | f"Request {req_id} has {num_prompt_tokens}" |
| 514 | " prompt tokens but has" |
| 515 | f" {len_prompt_logprobs} prompt logprobs." |
| 516 | ) |
| 517 | # First prompt logprob is None |
| 518 | first_plp_dict = prompt_logprobs[0] |
| 519 | assert first_plp_dict is None, ( |
| 520 | f"Request {req_id} first prompt logprob" |
| 521 | f" should be None but has following value" |
| 522 | f" instead: {first_plp_dict}" |
| 523 | ) |
| 524 | # Break out the reference prompt log prob value & |
| 525 | # logprob token id matrices for the whole prompt. |
| 526 | # Also break out the prompt token rank vector |
| 527 | ( |
| 528 | ref_prompt_logprob_toks, |
| 529 | ref_prompt_logprob_vals, |
| 530 | ref_prompt_token_ranks, |
| 531 | *_, |
| 532 | ) = ref_prompt_logprobs |
| 533 | for idx, (prompt_token, pos_logprob_dict) in enumerate( |
| 534 | zip(prompt_token_ids[1:], prompt_logprobs[1:]) |
| 535 | ): |
| 536 | # Break out the reference prompt log prob value |
| 537 | # vector, prompt logprob token id vector, and |
| 538 | # prompt token rank at the current position. |
| 539 | ( |
| 540 | ref_pos_prompt_logprob_toks, |
| 541 | ref_pos_prompt_logprob_vals, |
| 542 | ref_pos_prompt_token_rank, |
| 543 | ) = ( |
| 544 | ref_prompt_logprob_toks[idx, :], |
| 545 | ref_prompt_logprob_vals[idx, :], |
| 546 | ref_prompt_token_ranks[idx], |
| 547 | ) |
| 548 | |
| 549 | # For each position in the prompt sequence, |
| 550 | # ensure the actual prompt token is among the |
| 551 | # logprobs |
| 552 | assert prompt_token in pos_logprob_dict, ( |
| 553 | f"Prompt token {prompt_token} not present in logprob at index {idx}" |
| 554 | ) |
| 555 | # Validate number of prompt logprobs |
| 556 | num_plp_toks = len(pos_logprob_dict) |
| 557 | assert ( |
| 558 | num_plp_toks == num_prompt_logprobs |
| 559 | or num_plp_toks == num_prompt_logprobs + 1 |
| 560 | ), ( |
| 561 | "Valid numbers of prompt logprobs are" |
| 562 | f" {num_prompt_logprobs} or" |
| 563 | f" {num_prompt_logprobs + 1} but" |
| 564 | f" {num_plp_toks} logprobs found at" |
| 565 | f" position {idx}. Logprobs dict:" |
| 566 | f" {pos_logprob_dict}" |
| 567 | ) |
| 568 | |
| 569 | # Validate prompt token logprob rank |
| 570 | prmpt_tok_lp = pos_logprob_dict[prompt_token] |
| 571 | prmpt_tok_lp_rank = prmpt_tok_lp.rank |
| 572 | ref_prmpt_tok_lp_rank = ref_pos_prompt_token_rank |
| 573 | assert ref_prmpt_tok_lp_rank == prmpt_tok_lp_rank, ( |
| 574 | "Prompt token logprob rank" |
| 575 | f" {prmpt_tok_lp_rank} does not match" |
| 576 | " correct value" |
| 577 | f" {ref_prmpt_tok_lp_rank}" |
| 578 | f" in Logprob {prmpt_tok_lp}" |
| 579 | ) |
| 580 | |
| 581 | # Validate that the logprob processor yields |
| 582 | # the correct prompt log probs and valid |
| 583 | # rankings |
| 584 | rank_one_appears = False |
| 585 | for jdx in range(1, len(ref_pos_prompt_logprob_toks)): |
| 586 | # Iterate over the (logprob val,logprob tok id) |
| 587 | # pairs expected by the test fixture at this |
| 588 | # position in the completion. |
| 589 | ref_plp_val = float(ref_pos_prompt_logprob_vals[jdx]) |
| 590 | ref_tok_id = int(ref_pos_prompt_logprob_toks[jdx]) |
| 591 | assert ref_tok_id in pos_logprob_dict, ( |
| 592 | f"Expected token {ref_tok_id} to be" |
| 593 | f" in logprob dict but it is not." |
| 594 | ) |
| 595 | |
| 596 | # Extract actually-generated logprob |
| 597 | # info |
| 598 | plp = pos_logprob_dict[ref_tok_id] |
| 599 | plp_val = plp.logprob |
| 600 | plp_rank = plp.rank |
| 601 | |
| 602 | # A "top" (rank 1) logprob must be |
| 603 | # present |
| 604 | rank_one_appears = True if plp_rank == 1 else rank_one_appears |
| 605 | |
| 606 | # Rank must be >= 1 |
| 607 | assert plp_rank >= 1, ( |
| 608 | f"Logprob {plp} has invalid" |
| 609 | f" rank {plp_rank} < 1." |
| 610 | f" Logprob dict: {pos_logprob_dict}" |
| 611 | ) |
| 612 | |
| 613 | # Validate log probability |
| 614 | assert math.isclose(plp_val, ref_plp_val), ( |
| 615 | f"Token id {ref_tok_id} appears in logprobs dict" |
| 616 | f" at position {idx} in completion with log" |
| 617 | f" probability {plp_val} but {ref_plp_val} was" |
| 618 | f" expected. Logprob: {plp}" |
| 619 | ) |
| 620 | |
| 621 | assert rank_one_appears, ( |
| 622 | f"No Logprob has rank 1" |
| 623 | " in the following Logprob" |
| 624 | f" dict: {pos_logprob_dict}" |
| 625 | ) |
| 626 | |
| 627 | # Validate prompt logprob detokenization |
| 628 | for plp_tok in pos_logprob_dict: |
| 629 | # Confirm that prompt logprob decoded token matches |
| 630 | # the logprob token id at this sequence position |
| 631 | decoded_token = pos_logprob_dict[plp_tok].decoded_token |
| 632 | ref_decoded_token = _ref_convert_id_to_token(dtv.tokenizer, plp_tok) |
| 633 | |
| 634 | # With UTF-8 correction logic, tokens ending with "�" |
| 635 | # (incomplete byte sequences) are corrected to either |
| 636 | # empty string or proper UTF-8 characters |
| 637 | if ref_decoded_token.endswith("�"): |
| 638 | # Token needs UTF-8 correction |
| 639 | assert not decoded_token.endswith("�"), ( |
| 640 | f"Prompt logprob token id {plp_tok} decodes to" |
| 641 | f" '{ref_decoded_token}' (ends with replacement char)" |
| 642 | f" but corrected decoded token '{decoded_token}'" |
| 643 | f" still ends with replacement char" |
| 644 | f" (at position {idx}). UTF-8 correction should" |
| 645 | f" have removed it." |
| 646 | ) |
| 647 | else: |
| 648 | # No correction needed, should match exactly |
| 649 | assert decoded_token == ref_decoded_token, ( |
| 650 | f"Prompt logprob token id {plp_tok} decodes to" |
| 651 | f" {ref_decoded_token} but Logprob decoded" |
| 652 | f" token is {decoded_token} instead" |
| 653 | f" (at position {idx})" |
| 654 | ) |
| 655 | else: |
| 656 | # Prompt logprobs disabled for this request |
| 657 | assert prompt_logprobs is None |
| 658 | |
| 659 | |
| 660 | @pytest.mark.parametrize( |
| 661 | "request_output_kind", [RequestOutputKind.DELTA, RequestOutputKind.FINAL_ONLY] |
| 662 | ) |
| 663 | @pytest.mark.parametrize("num_sample_logprobs", [None, NUM_SAMPLE_LOGPROBS_UNDER_TEST]) |
| 664 | @pytest.mark.parametrize("num_prompt_logprobs", [None, NUM_PROMPT_LOGPROBS_UNDER_TEST]) |
| 665 | def test_logprobs_processor( |
| 666 | request_output_kind: RequestOutputKind, |
| 667 | num_sample_logprobs: int | None, |
| 668 | num_prompt_logprobs: int | None, |
| 669 | dummy_test_vectors, |
| 670 | ): |
| 671 | output_processor = OutputProcessor(dummy_test_vectors.tokenizer, log_stats=False) |
| 672 | |
| 673 | # Make N requests. |
| 674 | request_id_list = [ |
| 675 | f"request-{idx}" for idx in range(len(dummy_test_vectors.prompt_strings)) |
| 676 | ] |
| 677 | requests = [ |
| 678 | EngineCoreRequest( |
| 679 | request_id=request_id_list[idx] + "-int", |
| 680 | external_req_id=request_id_list[idx], |
| 681 | prompt_token_ids=prompt_tokens, |
| 682 | mm_features=None, |
| 683 | arrival_time=0, |
| 684 | lora_request=None, |
| 685 | cache_salt=None, |
| 686 | data_parallel_rank=None, |
| 687 | sampling_params=SamplingParams( |
| 688 | skip_special_tokens=False, |
| 689 | spaces_between_special_tokens=False, |
| 690 | output_kind=request_output_kind, |
| 691 | stop=[], |
| 692 | include_stop_str_in_output=False, |
| 693 | logprobs=num_sample_logprobs, |
| 694 | prompt_logprobs=num_prompt_logprobs, |
| 695 | ), |
| 696 | pooling_params=None, |
| 697 | ) |
| 698 | for idx, prompt_tokens in enumerate(dummy_test_vectors.prompt_tokens) |
| 699 | ] |
| 700 | |
| 701 | engine_core = MockEngineCore( |
| 702 | tokens_list=dummy_test_vectors.generation_tokens, |
| 703 | prompts_list=dummy_test_vectors.prompt_tokens, |
| 704 | generated_logprobs_raw=None |
| 705 | if num_sample_logprobs is None |
| 706 | else dummy_test_vectors.generation_logprobs, |
| 707 | prompt_logprobs_raw=None |
| 708 | if num_prompt_logprobs is None |
| 709 | else dummy_test_vectors.prompt_logprobs, |
| 710 | request_ids=[req.request_id for req in requests], |
| 711 | ) |
| 712 | |
| 713 | # Add requests to the detokenizer. |
| 714 | for request, prompt in zip(requests, dummy_test_vectors.prompt_strings): |
| 715 | output_processor.add_request(request, prompt) |
| 716 | |
| 717 | gen_tokens = {} |
| 718 | gen_logprobs = {} |
| 719 | gen_prompt_logprobs = {} |
| 720 | gen_cumulative_logprobs = {} |
| 721 | while True: |
| 722 | # Mock output from the EngineCore. |
| 723 | outputs = engine_core.get_outputs() |
| 724 | if len(outputs) == 0: |
| 725 | break |
| 726 | |
| 727 | # Step the logprobs processor. |
| 728 | processed_outputs = output_processor.process_outputs(outputs) |
| 729 | request_outputs = processed_outputs.request_outputs |
| 730 | requests_to_abort = processed_outputs.reqs_to_abort |
| 731 | assert len(requests_to_abort) == 0 |
| 732 | |
| 733 | # Update tracking. |
| 734 | for request_output in request_outputs: |
| 735 | request_id = request_output.request_id |
| 736 | new_tokens = request_output.outputs[0].token_ids |
| 737 | prompt_logprobs = request_output.prompt_logprobs |
| 738 | logprobs = request_output.outputs[0].logprobs |
| 739 | gen_cumulative_logprobs[request_id] = request_output.outputs[ |
| 740 | 0 |
| 741 | ].cumulative_logprob |
| 742 | if request_id not in gen_logprobs: |
| 743 | # Start tracking sample and prompt logprobs for this request |
| 744 | gen_tokens[request_id] = new_tokens |
| 745 | gen_logprobs[request_id] = logprobs |
| 746 | gen_prompt_logprobs[request_id] = prompt_logprobs |
| 747 | else: |
| 748 | # Extend logprobs tracker |
| 749 | gen_tokens[request_id].extend(new_tokens) |
| 750 | lp = gen_logprobs[request_id] |
| 751 | plp = gen_prompt_logprobs[request_id] |
| 752 | if lp: |
| 753 | lp.extend(logprobs) |
| 754 | if plp: |
| 755 | plp.extend(prompt_logprobs) |
| 756 | |
| 757 | # Confirmed tracked logprobs match what we expect |
| 758 | _validate_logprobs( |
| 759 | gen_tokens, |
| 760 | gen_logprobs, |
| 761 | gen_prompt_logprobs, |
| 762 | gen_cumulative_logprobs, |
| 763 | dummy_test_vectors, |
| 764 | request_id_list, |
| 765 | num_sample_logprobs, |
| 766 | num_prompt_logprobs, |
| 767 | ) |
| 768 | |
| 769 | assert output_processor.get_num_unfinished_requests() == 0 |
| 770 | assert not output_processor.has_unfinished_requests() |
| 771 | |
| 772 | |
| 773 | @pytest.mark.parametrize( |
| 774 | "include_stop_str_in_output,stop_token_type,ignore_eos,num_sample_logprobs", |
| 775 | [ |
| 776 | (False, "stop_token_ids", False, None), |
| 777 | (True, "stop_token_ids", False, None), |
| 778 | (False, "stop_token_ids", False, NUM_SAMPLE_LOGPROBS_UNDER_TEST), |
| 779 | (True, "stop_token_ids", False, NUM_SAMPLE_LOGPROBS_UNDER_TEST), |
| 780 | (False, "eos_token_id", False, None), |
| 781 | (True, "eos_token_id", False, None), |
| 782 | (False, "eos_token_id", True, None), |
| 783 | ], |
| 784 | ) |
| 785 | def test_stop_token( |
| 786 | include_stop_str_in_output: bool, |
| 787 | num_sample_logprobs: int | None, |
| 788 | stop_token_type: str, |
| 789 | ignore_eos: bool, |
| 790 | dummy_test_vectors, |
| 791 | ): |
| 792 | """Test output processor EOS/stop token handling. |
| 793 | |
| 794 | Send mock engine core request to mock engine core and pass core outputs |
| 795 | to output processor. Validate output processor tokens, text and |
| 796 | (if enabled) sample logprobs. Batch-size one. |
| 797 | |
| 798 | The test emulates a scenario where a model outputs text tokens followed |
| 799 | by two identical control tokens: |
| 800 | <token><token>...<token><control><control> |
| 801 | |
| 802 | If EOS is under test, the control tokens are EOS; otherwise, they are |
| 803 | some other token id. |
| 804 | |
| 805 | Test behavior: |
| 806 | |
| 807 | * If EOS is under test and `ignore_eos=True`, the detokenized string |
| 808 | should be <token><token>...<token><control><control> and the finish |
| 809 | reason should be "length" (i.e. no stop occurs) |
| 810 | |
| 811 | * else, if `include_stop_str_in_output==True`, the detokenized |
| 812 | string should be <token><token>...<token><control> and the finish |
| 813 | reason should be "stop" (i.e. first control token causes stop |
| 814 | and is represented in output text) |
| 815 | |
| 816 | * else, the detokenized string should be |
| 817 | <token><token>...<token> and the finish reason should be "stop" |
| 818 | (i.e. first control token causes stop but is not represented |
| 819 | in output text.) |
| 820 | |
| 821 | Note: some test details are tuned for meta-llama/Llama-3.2-1B, |
| 822 | another model should work only if the test is modified. |
| 823 | |
| 824 | Args: |
| 825 | include_stop_str_in_output: stop token str appears in output text |
| 826 | num_sample_logprobs: number of sample logprobs (`None` for no logprobs) |
| 827 | stop_token_type: "eos_token_id" for EOS, "stop_token_ids" for stop token |
| 828 | ignore_eos: if True, EOS stops are disabled |
| 829 | dummy_test_vectors: dummy engine core outputs and other data structures |
| 830 | |
| 831 | """ |
| 832 | model_id = dummy_test_vectors.tokenizer.name_or_path |
| 833 | if model_id != "meta-llama/Llama-3.2-1B": |
| 834 | raise AssertionError( |
| 835 | f"Test requires meta-llama/Llama-3.2-1B but {model_id} is in use." |
| 836 | ) |
| 837 | do_logprobs = num_sample_logprobs is not None |
| 838 | # EOS under test; if False, stop_token_ids under test |
| 839 | is_eos_test = stop_token_type == "eos_token_id" |
| 840 | # EOS under test but ignore_eos enabled |
| 841 | is_eos_ignore_test = is_eos_test and ignore_eos |
| 842 | eos_token_id = ( |
| 843 | dummy_test_vectors.tokenizer.eos_token_id if is_eos_test else None |
| 844 | ) # '<|end_of_text|>' |
| 845 | stop_token_ids = [128009] if not is_eos_test else None # '<|eot_id|>' |
| 846 | |
| 847 | output_processor = OutputProcessor(dummy_test_vectors.tokenizer, log_stats=False) |
| 848 | # Dummy engine core outputs, with control tokens suffixed to test stops |
| 849 | suffix_token = [eos_token_id] if is_eos_test else stop_token_ids |
| 850 | assert suffix_token is not None and isinstance(suffix_token[0], int) |
| 851 | generation_string = dummy_test_vectors.generation_strings[0] |
| 852 | generation_tokens = dummy_test_vectors.generation_tokens[0] + 2 * suffix_token |
| 853 | if do_logprobs: |
| 854 | generation_logprobs = dummy_test_vectors.generation_logprobs[0] + 2 * [ |
| 855 | dummy_test_vectors.generation_logprobs[0][-1] |
| 856 | ] |
| 857 | prompt_string = dummy_test_vectors.prompt_strings[0] |
| 858 | prompt_tokens = dummy_test_vectors.prompt_tokens[0] |
| 859 | |
| 860 | sampling_params = SamplingParams( |
| 861 | skip_special_tokens=False, |
| 862 | spaces_between_special_tokens=False, |
| 863 | output_kind=RequestOutputKind.DELTA, |
| 864 | stop=[], |
| 865 | stop_token_ids=stop_token_ids, |
| 866 | include_stop_str_in_output=include_stop_str_in_output, |
| 867 | logprobs=num_sample_logprobs, |
| 868 | prompt_logprobs=None, |
| 869 | ignore_eos=ignore_eos, |
| 870 | ) |
| 871 | sampling_params.update_from_generation_config({}, eos_token_id) |
| 872 | |
| 873 | # Make request. |
| 874 | request_id = "request-0" |
| 875 | request = EngineCoreRequest( |
| 876 | request_id=request_id, |
| 877 | external_req_id=request_id + "-ext", |
| 878 | prompt_token_ids=prompt_tokens, |
| 879 | mm_features=None, |
| 880 | arrival_time=0, |
| 881 | lora_request=None, |
| 882 | cache_salt=None, |
| 883 | data_parallel_rank=None, |
| 884 | sampling_params=sampling_params, |
| 885 | pooling_params=None, |
| 886 | ) |
| 887 | |
| 888 | engine_core = MockEngineCore( |
| 889 | tokens_list=[generation_tokens], |
| 890 | prompts_list=dummy_test_vectors.prompt_tokens, |
| 891 | generated_logprobs_raw=[generation_logprobs] if do_logprobs else None, |
| 892 | prompt_logprobs_raw=None, |
| 893 | eos_token_id=sampling_params.eos_token_id, |
| 894 | stop_token_ids=sampling_params.stop_token_ids, |
| 895 | request_ids=[request.request_id], |
| 896 | ) |
| 897 | |
| 898 | # Add request to the detokenizer. |
| 899 | output_processor.add_request(request, prompt_string) |
| 900 | |
| 901 | # Loop over engine core steps; run output processor |
| 902 | gen_string = "" |
| 903 | gen_tokens = [] |
| 904 | gen_logprobs = [] |
| 905 | while True: |
| 906 | # Mock output from the EngineCore. |
| 907 | outputs = engine_core.get_outputs() |
| 908 | if len(outputs) == 0: |
| 909 | break |
| 910 | |
| 911 | # Step the Detokenizer. |
| 912 | processed_outputs = output_processor.process_outputs(outputs) |
| 913 | request_outputs = processed_outputs.request_outputs |
| 914 | assert len(request_outputs) == 1 |
| 915 | # Stop token does not rely on abort |
| 916 | assert not processed_outputs.reqs_to_abort |
| 917 | |
| 918 | # Update tracking. |
| 919 | request_output = request_outputs[0] |
| 920 | if request_output.finished: |
| 921 | finish_reason = "length" if is_eos_ignore_test else "stop" |
| 922 | assert request_output.outputs[0].finish_reason == finish_reason |
| 923 | |
| 924 | gen_string += request_output.outputs[0].text |
| 925 | gen_tokens.extend(request_output.outputs[0].token_ids) |
| 926 | if do_logprobs: |
| 927 | gen_logprobs.extend(request_output.outputs[0].logprobs) |
| 928 | |
| 929 | # Validate generated text |
| 930 | control_token = "<|end_of_text|>" if is_eos_test else "<|eot_id|>" |
| 931 | if is_eos_ignore_test: |
| 932 | # Length-based stop; expect full string |
| 933 | ref_str = generation_string + 2 * control_token |
| 934 | elif include_stop_str_in_output: |
| 935 | # Stop token triggered; include in output |
| 936 | ref_str = generation_string + control_token |
| 937 | else: |
| 938 | # Stop token triggered but not in output |
| 939 | ref_str = generation_string |
| 940 | assert gen_string == ref_str, f"{gen_string=}, {ref_str=}" |
| 941 | |
| 942 | if do_logprobs: |
| 943 | # Validate number of sample logprobs |
| 944 | num_tokens = len(gen_tokens) |
| 945 | num_logprobs = len(gen_logprobs) |
| 946 | assert num_tokens == num_logprobs, ( |
| 947 | f"Token count ({num_tokens}) != logprobs count ({num_logprobs})" |
| 948 | ) |
| 949 | |
| 950 | # Check requests are finished |
| 951 | assert output_processor.get_num_unfinished_requests() == 0 |
| 952 | assert not output_processor.has_unfinished_requests() |
| 953 | |
| 954 | |
| 955 | @pytest.mark.parametrize("include_stop_str_in_output", [True, False]) |
| 956 | @pytest.mark.parametrize("num_sample_logprobs", [None, NUM_SAMPLE_LOGPROBS_UNDER_TEST]) |
| 957 | def test_stop_string( |
| 958 | include_stop_str_in_output: bool, |
| 959 | num_sample_logprobs: int | None, |
| 960 | dummy_test_vectors, |
| 961 | ): |
| 962 | output_processor = OutputProcessor(dummy_test_vectors.tokenizer, log_stats=False) |
| 963 | |
| 964 | # Make N requests. |
| 965 | request_id_list = [ |
| 966 | f"request-{idx}" for idx in range(len(dummy_test_vectors.prompt_strings)) |
| 967 | ] |
| 968 | requests = [ |
| 969 | EngineCoreRequest( |
| 970 | request_id=request_id_list[idx] + "-int", |
| 971 | external_req_id=request_id_list[idx], |
| 972 | prompt_token_ids=prompt_tokens, |
| 973 | mm_features=None, |
| 974 | arrival_time=0, |
| 975 | lora_request=None, |
| 976 | cache_salt=None, |
| 977 | data_parallel_rank=None, |
| 978 | sampling_params=SamplingParams( |
| 979 | skip_special_tokens=False, |
| 980 | spaces_between_special_tokens=False, |
| 981 | output_kind=RequestOutputKind.DELTA, |
| 982 | stop=STOP_STRINGS, |
| 983 | include_stop_str_in_output=include_stop_str_in_output, |
| 984 | logprobs=num_sample_logprobs, |
| 985 | prompt_logprobs=None, |
| 986 | ), |
| 987 | pooling_params=None, |
| 988 | ) |
| 989 | for idx, prompt_tokens in enumerate(dummy_test_vectors.prompt_tokens) |
| 990 | ] |
| 991 | |
| 992 | engine_core = MockEngineCore( |
| 993 | tokens_list=dummy_test_vectors.generation_tokens, |
| 994 | prompts_list=dummy_test_vectors.prompt_tokens, |
| 995 | generated_logprobs_raw=dummy_test_vectors.generation_logprobs |
| 996 | if num_sample_logprobs |
| 997 | else None, |
| 998 | prompt_logprobs_raw=None, |
| 999 | request_ids=[req.request_id for req in requests], |
| 1000 | ) |
| 1001 | |
| 1002 | # Add requests to the detokenizer. |
| 1003 | for request, prompt in zip(requests, dummy_test_vectors.prompt_strings): |
| 1004 | output_processor.add_request(request, prompt) |
| 1005 | |
| 1006 | gen_strings = {} |
| 1007 | gen_tokens = {} |
| 1008 | gen_logprobs = {} |
| 1009 | gen_prompt_logprobs = {} |
| 1010 | gen_cumulative_logprobs = {} |
| 1011 | aborted = [] |
| 1012 | while True: |
| 1013 | # Mock output from the EngineCore. |
| 1014 | outputs = engine_core.get_outputs() |
| 1015 | if len(outputs) == 0: |
| 1016 | break |
| 1017 | |
| 1018 | # Step the Detokenizer. |
| 1019 | processed_outputs = output_processor.process_outputs(outputs) |
| 1020 | request_outputs = processed_outputs.request_outputs |
| 1021 | requests_to_abort = processed_outputs.reqs_to_abort |
| 1022 | for request_output in request_outputs: |
| 1023 | # If aborted, we should not get a request output. |
| 1024 | assert request_output.request_id not in aborted |
| 1025 | aborted.extend(requests_to_abort) |
| 1026 | |
| 1027 | # Update tracking. |
| 1028 | for request_output in request_outputs: |
| 1029 | if request_output.finished: |
| 1030 | assert request_output.outputs[0].finish_reason == "stop" |
| 1031 | |
| 1032 | request_id = request_output.request_id |
| 1033 | new_text = request_output.outputs[0].text |
| 1034 | new_tokens = request_output.outputs[0].token_ids |
| 1035 | prompt_logprobs = request_output.prompt_logprobs |
| 1036 | logprobs = request_output.outputs[0].logprobs |
| 1037 | gen_cumulative_logprobs[request_id] = request_output.outputs[ |
| 1038 | 0 |
| 1039 | ].cumulative_logprob |
| 1040 | if request_id not in gen_strings: |
| 1041 | gen_strings[request_id] = new_text |
| 1042 | gen_tokens[request_id] = new_tokens |
| 1043 | gen_logprobs[request_id] = logprobs |
| 1044 | gen_prompt_logprobs[request_id] = prompt_logprobs |
| 1045 | else: |
| 1046 | gen_strings[request_id] += new_text |
| 1047 | gen_tokens[request_id].extend(new_tokens) |
| 1048 | lp = gen_logprobs[request_id] |
| 1049 | plp = gen_prompt_logprobs[request_id] |
| 1050 | if lp: |
| 1051 | lp.extend(logprobs) |
| 1052 | if plp: |
| 1053 | plp.extend(prompt_logprobs) |
| 1054 | |
| 1055 | # Confirmed tracked values matches what we expected. |
| 1056 | for idx, (ref_gen_str, stop_str) in enumerate( |
| 1057 | zip(dummy_test_vectors.generation_strings, STOP_STRINGS) |
| 1058 | ): |
| 1059 | # Request should be aborted (check internal ID in abort list). |
| 1060 | internal_request_id = f"request-{idx}-int" |
| 1061 | assert internal_request_id in aborted |
| 1062 | |
| 1063 | # Use external ID for collecting outputs |
| 1064 | request_id = f"request-{idx}" |
| 1065 | |
| 1066 | # Collected values that were generated. |
| 1067 | gen_str = gen_strings[request_id] |
| 1068 | |
| 1069 | # Construct reference strings. |
| 1070 | stop_str_idx = ref_gen_str.find(stop_str) |
| 1071 | ref_str_exc_stop = ref_gen_str[:stop_str_idx] |
| 1072 | ref_str_inc_stop = ref_gen_str[:stop_str_idx] + stop_str |
| 1073 | |
| 1074 | if include_stop_str_in_output: |
| 1075 | assert gen_str == ref_str_inc_stop, f"{gen_str=}, {ref_str_inc_stop=}" |
| 1076 | else: |
| 1077 | assert gen_str == ref_str_exc_stop, f"{gen_str=}, {ref_str_exc_stop=}" |
| 1078 | |
| 1079 | # Confirmed tracked logprobs match what we expect |
| 1080 | _validate_logprobs( |
| 1081 | gen_tokens, |
| 1082 | gen_logprobs, |
| 1083 | gen_prompt_logprobs, |
| 1084 | gen_cumulative_logprobs, |
| 1085 | dummy_test_vectors, |
| 1086 | request_id_list, |
| 1087 | num_sample_logprobs, |
| 1088 | None, |
| 1089 | ) |
| 1090 | |
| 1091 | assert output_processor.get_num_unfinished_requests() == 0 |
| 1092 | assert not output_processor.has_unfinished_requests() |
| 1093 | |
| 1094 | |
| 1095 | def test_iteration_stats(dummy_test_vectors): |
| 1096 | output_processor = OutputProcessor(dummy_test_vectors.tokenizer, log_stats=True) |
| 1097 | engine_core_timestamp = time.monotonic() |
| 1098 | |
| 1099 | # Make N requests. |
| 1100 | requests = [ |
| 1101 | EngineCoreRequest( |
| 1102 | request_id=f"request-{idx}", |
| 1103 | external_req_id=f"request-{idx}-ext", |
| 1104 | prompt_token_ids=prompt_tokens, |
| 1105 | mm_features=None, |
| 1106 | arrival_time=0, |
| 1107 | lora_request=None, |
| 1108 | cache_salt=None, |
| 1109 | data_parallel_rank=None, |
| 1110 | sampling_params=SamplingParams(), |
| 1111 | pooling_params=None, |
| 1112 | ) |
| 1113 | for idx, prompt_tokens in enumerate(dummy_test_vectors.prompt_tokens) |
| 1114 | ] |
| 1115 | |
| 1116 | engine_core = MockEngineCore( |
| 1117 | dummy_test_vectors.generation_tokens, |
| 1118 | dummy_test_vectors.prompt_tokens, |
| 1119 | request_ids=[req.request_id for req in requests], |
| 1120 | ) |
| 1121 | |
| 1122 | # Add all requests except one to the OutputProcessor. |
| 1123 | num_active = len(dummy_test_vectors.generation_tokens) - 1 |
| 1124 | for request in requests[:num_active]: |
| 1125 | output_processor.add_request(request, None) |
| 1126 | inactive_request = requests[num_active] |
| 1127 | |
| 1128 | # First iteration has 2 prefills. |
| 1129 | outputs = engine_core.get_outputs(num_active) |
| 1130 | iteration_stats = IterationStats() |
| 1131 | output_processor.process_outputs(outputs, engine_core_timestamp, iteration_stats) |
| 1132 | total_prompt_tokens = sum( |
| 1133 | [ |
| 1134 | len(prompt_tokens) |
| 1135 | for prompt_tokens in dummy_test_vectors.prompt_tokens[:num_active] |
| 1136 | ] |
| 1137 | ) |
| 1138 | |
| 1139 | assert iteration_stats.num_prompt_tokens == total_prompt_tokens |
| 1140 | assert iteration_stats.num_generation_tokens == num_active |
| 1141 | |
| 1142 | # Just decodes in this step. |
| 1143 | outputs = engine_core.get_outputs(num_active) |
| 1144 | iteration_stats = IterationStats() |
| 1145 | output_processor.process_outputs(outputs, engine_core_timestamp, iteration_stats) |
| 1146 | |
| 1147 | assert iteration_stats.num_prompt_tokens == 0 |
| 1148 | assert iteration_stats.num_generation_tokens == num_active |
| 1149 | |
| 1150 | # Add a new request - prefill and 2 decodes in this step. |
| 1151 | output_processor.add_request(inactive_request, None) |
| 1152 | num_active += 1 |
| 1153 | outputs = engine_core.get_outputs(num_active) |
| 1154 | iteration_stats = IterationStats() |
| 1155 | output_processor.process_outputs(outputs, engine_core_timestamp, iteration_stats) |
| 1156 | total_prompt_tokens = len(dummy_test_vectors.prompt_tokens[num_active - 1]) |
| 1157 | |
| 1158 | assert iteration_stats.num_prompt_tokens == total_prompt_tokens |
| 1159 | assert iteration_stats.num_generation_tokens == num_active |
| 1160 | |
| 1161 | # Just decodes in this step. |
| 1162 | outputs = engine_core.get_outputs(num_active) |
| 1163 | iteration_stats = IterationStats() |
| 1164 | output_processor.process_outputs(outputs, engine_core_timestamp, iteration_stats) |
| 1165 | |
| 1166 | assert iteration_stats.num_prompt_tokens == 0 |
| 1167 | assert iteration_stats.num_generation_tokens == num_active |
| 1168 | |
| 1169 | |
| 1170 | @pytest.mark.parametrize("log_stats", [True, False]) |
| 1171 | def test_lora_request_tracking(log_stats: bool, dummy_test_vectors): |
| 1172 | """Test LoRA request lifecycle tracking through waiting -> running -> finished.""" |
| 1173 | output_processor = OutputProcessor( |
| 1174 | dummy_test_vectors.tokenizer, log_stats=log_stats |
| 1175 | ) |
| 1176 | engine_core_timestamp = time.monotonic() |
| 1177 | |
| 1178 | # Create LoRA requests |
| 1179 | lora1 = LoRARequest(lora_name="lora-1", lora_int_id=1, lora_path="/path/to/lora1") |
| 1180 | lora2 = LoRARequest(lora_name="lora-2", lora_int_id=2, lora_path="/path/to/lora2") |
| 1181 | |
| 1182 | # Create requests with different LoRA adapters: |
| 1183 | # - request-0: lora-1 |
| 1184 | # - request-1: lora-2 |
| 1185 | # - request-2: None (no LoRA) |
| 1186 | lora_assignments = [lora1, lora2, None] |
| 1187 | requests = [ |
| 1188 | EngineCoreRequest( |
| 1189 | request_id=f"request-{idx}-int", |
| 1190 | external_req_id=f"request-{idx}", |
| 1191 | prompt_token_ids=prompt_tokens, |
| 1192 | mm_features=None, |
| 1193 | arrival_time=0, |
| 1194 | lora_request=lora_assignments[idx], |
| 1195 | cache_salt=None, |
| 1196 | data_parallel_rank=None, |
| 1197 | sampling_params=SamplingParams(), |
| 1198 | pooling_params=None, |
| 1199 | ) |
| 1200 | for idx, prompt_tokens in enumerate(dummy_test_vectors.prompt_tokens) |
| 1201 | ] |
| 1202 | |
| 1203 | engine_core = MockEngineCore( |
| 1204 | dummy_test_vectors.generation_tokens, |
| 1205 | dummy_test_vectors.prompt_tokens, |
| 1206 | request_ids=[req.request_id for req in requests], |
| 1207 | ) |
| 1208 | |
| 1209 | # Add all requests to the OutputProcessor |
| 1210 | for request in requests: |
| 1211 | output_processor.add_request(request, None) |
| 1212 | |
| 1213 | # First iteration: process outputs with QUEUED events |
| 1214 | outputs = EngineCoreOutputs( |
| 1215 | outputs=engine_core.get_outputs(), scheduler_stats=SchedulerStats() |
| 1216 | ) |
| 1217 | for output in outputs.outputs: |
| 1218 | output.events = [ |
| 1219 | EngineCoreEvent.new_event(EngineCoreEventType.QUEUED, engine_core_timestamp) |
| 1220 | ] |
| 1221 | |
| 1222 | iteration_stats = IterationStats() if log_stats else None |
| 1223 | output_processor.process_outputs( |
| 1224 | outputs.outputs, engine_core_timestamp, iteration_stats |
| 1225 | ) |
| 1226 | output_processor.update_scheduler_stats(outputs.scheduler_stats) |
| 1227 | |
| 1228 | if log_stats: |
| 1229 | # Verify waiting counts |
| 1230 | assert outputs.scheduler_stats.waiting_lora_adapters.get("lora-1") == 1 |
| 1231 | assert outputs.scheduler_stats.waiting_lora_adapters.get("lora-2") == 1 |
| 1232 | assert outputs.scheduler_stats.running_lora_adapters.get("lora-1") == 0 |
| 1233 | assert outputs.scheduler_stats.running_lora_adapters.get("lora-2") == 0 |
| 1234 | # Verify internal state |
| 1235 | assert len(output_processor.lora_states.requests) == 2 |
| 1236 | assert "lora-1" in output_processor.lora_states.requests |
| 1237 | assert "lora-2" in output_processor.lora_states.requests |
| 1238 | else: |
| 1239 | # When log_stats=False, no tracking should occur |
| 1240 | assert iteration_stats is None |
| 1241 | assert len(output_processor.lora_states.requests) == 0 |
| 1242 | |
| 1243 | # Second iteration: process outputs with SCHEDULED events |
| 1244 | outputs = EngineCoreOutputs( |
| 1245 | outputs=engine_core.get_outputs(), scheduler_stats=SchedulerStats() |
| 1246 | ) |
| 1247 | for output in outputs.outputs: |
| 1248 | output.events = [ |
| 1249 | EngineCoreEvent.new_event( |
| 1250 | EngineCoreEventType.SCHEDULED, engine_core_timestamp |
| 1251 | ) |
| 1252 | ] |
| 1253 | |
| 1254 | iteration_stats = IterationStats() if log_stats else None |
| 1255 | output_processor.process_outputs( |
| 1256 | outputs.outputs, engine_core_timestamp, iteration_stats |
| 1257 | ) |
| 1258 | output_processor.update_scheduler_stats(outputs.scheduler_stats) |
| 1259 | |
| 1260 | if log_stats: |
| 1261 | # Verify running counts |
| 1262 | assert outputs.scheduler_stats.waiting_lora_adapters.get("lora-1") == 0 |
| 1263 | assert outputs.scheduler_stats.waiting_lora_adapters.get("lora-2") == 0 |
| 1264 | assert outputs.scheduler_stats.running_lora_adapters.get("lora-1") == 1 |
| 1265 | assert outputs.scheduler_stats.running_lora_adapters.get("lora-2") == 1 |
| 1266 | else: |
| 1267 | assert iteration_stats is None |
| 1268 | assert len(output_processor.lora_states.requests) == 0 |
| 1269 | |
| 1270 | # Third iteration: finish request-0 (lora-1) |
| 1271 | outputs = EngineCoreOutputs( |
| 1272 | outputs=engine_core.get_outputs(), scheduler_stats=SchedulerStats() |
| 1273 | ) |
| 1274 | # Find and mark request-0-int as finished (it uses lora-1) |
| 1275 | for output in outputs.outputs: |
| 1276 | if output.request_id == "request-0-int": |
| 1277 | output.finish_reason = FinishReason.LENGTH |
| 1278 | break |
| 1279 | |
| 1280 | iteration_stats = IterationStats() if log_stats else None |
| 1281 | output_processor.process_outputs( |
| 1282 | outputs.outputs, engine_core_timestamp, iteration_stats |
| 1283 | ) |
| 1284 | output_processor.update_scheduler_stats(outputs.scheduler_stats) |
| 1285 | |
| 1286 | if log_stats: |
| 1287 | # lora-1 should be removed since no requests remain |
| 1288 | assert "lora-1" not in output_processor.lora_states.requests |
| 1289 | # lora-2 should still be running |
| 1290 | assert outputs.scheduler_stats.running_lora_adapters.get("lora-2") == 1 |
| 1291 | assert len(output_processor.lora_states.requests) == 1 |
| 1292 | else: |
| 1293 | assert len(output_processor.lora_states.requests) == 0 |
| 1294 | |
| 1295 | # Fourth iteration: finish request-1 (lora-2) |
| 1296 | outputs = EngineCoreOutputs( |
| 1297 | outputs=engine_core.get_outputs(), scheduler_stats=SchedulerStats() |
| 1298 | ) |
| 1299 | # Find and mark request-1-int as finished (it uses lora-2) |
| 1300 | for output in outputs.outputs: |
| 1301 | if output.request_id == "request-1-int": |
| 1302 | output.finish_reason = FinishReason.LENGTH |
| 1303 | break |
| 1304 | |
| 1305 | iteration_stats = IterationStats() if log_stats else None |
| 1306 | output_processor.process_outputs( |
| 1307 | outputs.outputs, engine_core_timestamp, iteration_stats |
| 1308 | ) |
| 1309 | output_processor.update_scheduler_stats(outputs.scheduler_stats) |
| 1310 | |
| 1311 | if log_stats: |
| 1312 | # lora-2 should be removed since no requests remain |
| 1313 | assert "lora-2" not in output_processor.lora_states.requests |
| 1314 | assert len(outputs.scheduler_stats.running_lora_adapters) == 0 |
| 1315 | assert len(output_processor.lora_states.requests) == 0 |
| 1316 | else: |
| 1317 | assert len(output_processor.lora_states.requests) == 0 |
| 1318 | |
| 1319 | # Finish the last request (no LoRA) |
| 1320 | outputs = EngineCoreOutputs( |
| 1321 | outputs=engine_core.get_outputs(), scheduler_stats=SchedulerStats() |
| 1322 | ) |
| 1323 | # Find and mark request-2-int as finished (it has no LoRA) |
| 1324 | for output in outputs.outputs: |
| 1325 | if output.request_id == "request-2-int": |
| 1326 | output.finish_reason = FinishReason.LENGTH |
| 1327 | break |
| 1328 | |
| 1329 | iteration_stats = IterationStats() if log_stats else None |
| 1330 | output_processor.process_outputs( |
| 1331 | outputs.outputs, engine_core_timestamp, iteration_stats |
| 1332 | ) |
| 1333 | output_processor.update_scheduler_stats(outputs.scheduler_stats) |
| 1334 | |
| 1335 | # Verify all requests are finished |
| 1336 | assert output_processor.get_num_unfinished_requests() == 0 |
| 1337 | |
| 1338 | |
| 1339 | @pytest.mark.asyncio |
| 1340 | async def test_request_output_collector(): |
| 1341 | NUM_REQS = 3 |
| 1342 | TEXT = "a" |
| 1343 | routed_experts = np.arange(12, dtype=np.uint8).reshape(2, 3, 2) |
| 1344 | |
| 1345 | def make_outputs() -> list[RequestOutput]: |
| 1346 | return [ |
| 1347 | RequestOutput( |
| 1348 | request_id="my-request-id", |
| 1349 | prompt=None, |
| 1350 | prompt_token_ids=[1, 2, 3], |
| 1351 | prompt_logprobs=None, |
| 1352 | outputs=[ |
| 1353 | CompletionOutput( |
| 1354 | index=0, |
| 1355 | text=TEXT, |
| 1356 | token_ids=[idx], |
| 1357 | cumulative_logprob=(idx + 1 * 1.0), |
| 1358 | logprobs=[{"a": idx, "b": idx}], |
| 1359 | routed_experts=( |
| 1360 | routed_experts if idx == NUM_REQS - 1 else None |
| 1361 | ), |
| 1362 | finish_reason="length" if (idx == NUM_REQS - 1) else None, |
| 1363 | ) |
| 1364 | ], |
| 1365 | finished=(idx == NUM_REQS - 1), |
| 1366 | ) |
| 1367 | for idx in range(NUM_REQS) |
| 1368 | ] |
| 1369 | |
| 1370 | collector = RequestOutputCollector( |
| 1371 | RequestOutputKind.DELTA, request_id="my-request-id-int" |
| 1372 | ) |
| 1373 | |
| 1374 | # CASE 1: Put then get. |
| 1375 | outputs = make_outputs() |
| 1376 | collector.put(outputs[0]) |
| 1377 | output = await collector.get() |
| 1378 | assert not collector.ready.is_set() |
| 1379 | assert collector.output is None |
| 1380 | assert output.outputs[0].text == "a" |
| 1381 | assert output.outputs[0].token_ids == [0] |
| 1382 | |
| 1383 | # CASE 2: 2 puts then get. |
| 1384 | num_to_put = 2 |
| 1385 | outputs = make_outputs() |
| 1386 | for i in range(num_to_put): |
| 1387 | collector.put(outputs[i]) |
| 1388 | output = await collector.get() |
| 1389 | assert not collector.ready.is_set() |
| 1390 | assert collector.output is None |
| 1391 | |
| 1392 | assert not output.finished |
| 1393 | # Text, token_ids, and logprobs should get merged. |
| 1394 | assert output.outputs[0].text == TEXT * num_to_put |
| 1395 | for tok_0, tok_1 in zip(output.outputs[0].token_ids, list(range(num_to_put))): |
| 1396 | assert tok_0 == tok_1 |
| 1397 | assert len(output.outputs[0].logprobs) == num_to_put |
| 1398 | |
| 1399 | # Cumulative logprobs should be the last one. |
| 1400 | cumulative_logprob_expected = 1.0 * num_to_put |
| 1401 | assert output.outputs[0].cumulative_logprob == cumulative_logprob_expected |
| 1402 | |
| 1403 | # CASE 3: Put all 3 (including a finished). |
| 1404 | num_to_put = 3 |
| 1405 | outputs = make_outputs() |
| 1406 | for i in range(num_to_put): |
| 1407 | collector.put(outputs[i]) |
| 1408 | output = await collector.get() |
| 1409 | assert not collector.ready.is_set() |
| 1410 | assert collector.output is None |
| 1411 | |
| 1412 | assert output.finished |
| 1413 | assert output.outputs[0].finish_reason == "length" |
| 1414 | np.testing.assert_array_equal(output.outputs[0].routed_experts, routed_experts) |
| 1415 | # Text, token_ids, and logprobs should get merged. |
| 1416 | assert output.outputs[0].text == TEXT * num_to_put |
| 1417 | for tok_0, tok_1 in zip(output.outputs[0].token_ids, list(range(num_to_put))): |
| 1418 | assert tok_0 == tok_1 |
| 1419 | assert len(output.outputs[0].logprobs) == num_to_put |
| 1420 | |
| 1421 | # Cumulative logprobs should be the last one. |
| 1422 | cumulative_logprob_expected = 1.0 * num_to_put |
| 1423 | assert output.outputs[0].cumulative_logprob == cumulative_logprob_expected |
| 1424 | |
| 1425 | |
| 1426 | def test_routed_experts_are_accumulated_until_finish(): |
| 1427 | state = RequestState( |
| 1428 | request_id="request-int", |
| 1429 | external_req_id="request", |
| 1430 | parent_req=None, |
| 1431 | request_index=0, |
| 1432 | lora_request=None, |
| 1433 | output_kind=RequestOutputKind.DELTA, |
| 1434 | prompt="prompt", |
| 1435 | prompt_token_ids=[1, 2, 3, 4], |
| 1436 | prompt_embeds=None, |
| 1437 | logprobs_processor=Mock( |
| 1438 | logprobs=None, |
| 1439 | cumulative_logprob=0.0, |
| 1440 | pop_prompt_logprobs=Mock(return_value=None), |
| 1441 | ), |
| 1442 | detokenizer=Mock(get_next_output_text=Mock(return_value="")), |
| 1443 | max_tokens_param=2, |
| 1444 | arrival_time=0.0, |
| 1445 | queue=None, |
| 1446 | log_stats=False, |
| 1447 | stream_interval=1, |
| 1448 | ) |
| 1449 | prompt_chunk = np.arange(24, dtype=np.uint8).reshape(4, 3, 2) |
| 1450 | decode_chunk = np.arange(12, dtype=np.uint8).reshape(2, 3, 2) + 24 |
| 1451 | state.routed_experts_chunks.append(prompt_chunk) |
| 1452 | |
| 1453 | partial = state.make_request_output([5], None, None, None) |
| 1454 | assert partial is not None |
| 1455 | assert partial.outputs[0].routed_experts is None |
| 1456 | |
| 1457 | state.routed_experts_chunks.append(decode_chunk) |
| 1458 | finished = state.make_request_output([6], None, FinishReason.LENGTH, None) |
| 1459 | assert finished is not None |
| 1460 | np.testing.assert_array_equal( |
| 1461 | finished.outputs[0].routed_experts, |
| 1462 | np.concatenate((prompt_chunk, decode_chunk)), |
| 1463 | ) |
| 1464 | |
| 1465 | |
| 1466 | @pytest.mark.asyncio |
| 1467 | async def test_cumulative_output_collector_n(): |
| 1468 | """Test collector correctly handles multiple outputs by index.""" |
| 1469 | collector = RequestOutputCollector( |
| 1470 | RequestOutputKind.CUMULATIVE, request_id="my-request-id-int" |
| 1471 | ) |
| 1472 | outputs = [ |
| 1473 | RequestOutput( |
| 1474 | request_id="my-request-id", |
| 1475 | prompt=None, |
| 1476 | prompt_token_ids=[1, 2, 3], |
| 1477 | prompt_logprobs=None, |
| 1478 | outputs=[ |
| 1479 | CompletionOutput( |
| 1480 | index=0, |
| 1481 | text="a", |
| 1482 | token_ids=[0], |
| 1483 | cumulative_logprob=None, |
| 1484 | logprobs=None, |
| 1485 | finish_reason=None, |
| 1486 | ), |
| 1487 | CompletionOutput( |
| 1488 | index=1, |
| 1489 | text="b", |
| 1490 | token_ids=[1], |
| 1491 | cumulative_logprob=None, |
| 1492 | logprobs=None, |
| 1493 | finish_reason=None, |
| 1494 | ), |
| 1495 | ], |
| 1496 | finished=False, |
| 1497 | ), |
| 1498 | RequestOutput( |
| 1499 | request_id="my-request-id", |
| 1500 | prompt=None, |
| 1501 | prompt_token_ids=[1, 2, 3], |
| 1502 | prompt_logprobs=None, |
| 1503 | outputs=[ |
| 1504 | CompletionOutput( |
| 1505 | index=0, |
| 1506 | text="ab", |
| 1507 | token_ids=[0, 1], |
| 1508 | cumulative_logprob=None, |
| 1509 | logprobs=None, |
| 1510 | finish_reason=None, |
| 1511 | ), |
| 1512 | CompletionOutput( |
| 1513 | index=2, |
| 1514 | text="c", |
| 1515 | token_ids=[2], |
| 1516 | cumulative_logprob=None, |
| 1517 | logprobs=None, |
| 1518 | finish_reason=None, |
| 1519 | ), |
| 1520 | ], |
| 1521 | finished=False, |
| 1522 | ), |
| 1523 | ] |
| 1524 | for output in outputs: |
| 1525 | collector.put(output) |
| 1526 | |
| 1527 | # Get the output and check that the text and token_ids are correct. |
| 1528 | result = await collector.get() |
| 1529 | # We are expecting |
| 1530 | # [{index: 0, text: "ab"}, {index: 1, text: "b"}, {index: 2, text: "c"}] |
| 1531 | assert len(result.outputs) == 3 |
| 1532 | # First is the one where index is 0 |
| 1533 | first = [k for k in result.outputs if k.index == 0] |
| 1534 | assert len(first) == 1 |
| 1535 | assert first[0].text == "ab" |
| 1536 | |
| 1537 | # Second is the one where index is 1 |
| 1538 | second = [k for k in result.outputs if k.index == 1] |
| 1539 | assert len(second) == 1 |
| 1540 | assert second[0].text == "b" |
| 1541 | assert second[0].token_ids == [1] |
| 1542 | |
| 1543 | # Third is the one where index is 2 |
| 1544 | third = [k for k in result.outputs if k.index == 2] |
| 1545 | assert len(third) == 1 |
| 1546 | assert third[0].text == "c" |
| 1547 | |
| 1548 | |
| 1549 | @pytest.mark.parametrize("runner", ["generate", "pooling"]) |
| 1550 | @pytest.mark.parametrize("abort_by", ["internal", "external"]) |
| 1551 | def test_abort_requests(runner: str, abort_by: str, dummy_test_vectors): |
| 1552 | output_processor = OutputProcessor(dummy_test_vectors.tokenizer, log_stats=True) |
| 1553 | requests = [ |
| 1554 | EngineCoreRequest( |
| 1555 | request_id=f"request-{idx}", |
| 1556 | external_req_id=f"external-{idx}", |
| 1557 | prompt_token_ids=prompt_tokens, |
| 1558 | mm_features=None, |
| 1559 | arrival_time=0, |
| 1560 | lora_request=None, |
| 1561 | cache_salt=None, |
| 1562 | data_parallel_rank=None, |
| 1563 | sampling_params=SamplingParams() if runner == "generate" else None, |
| 1564 | pooling_params=PoolingParams(task="embed") if runner == "pooling" else None, |
| 1565 | ) |
| 1566 | for idx, prompt_tokens in enumerate(dummy_test_vectors.prompt_tokens) |
| 1567 | ] |
| 1568 | |
| 1569 | for request in requests: |
| 1570 | if runner == "generate": |
| 1571 | output_kind = request.sampling_params.output_kind |
| 1572 | else: |
| 1573 | output_kind = request.pooling_params.output_kind |
| 1574 | queue = RequestOutputCollector( |
| 1575 | output_kind=output_kind, request_id=request.request_id |
| 1576 | ) |
| 1577 | output_processor.add_request(request, None, queue=queue) |
| 1578 | |
| 1579 | for request in requests: |
| 1580 | if abort_by == "internal": |
| 1581 | output_processor.abort_requests([request.request_id], internal=True) |
| 1582 | else: |
| 1583 | output_processor.abort_requests([request.external_req_id], internal=False) |
| 1584 | |
| 1585 | |
| 1586 | @pytest.mark.parametrize("output_kind", list(RequestOutputKind)) |
| 1587 | def test_sampling_masks_follow_output_kind(output_kind): |
| 1588 | state = RequestState.__new__(RequestState) |
| 1589 | state.detokenizer = MagicMock() |
| 1590 | state.detokenizer.get_next_output_text.return_value = "" |
| 1591 | state.logprobs_processor = MagicMock() |
| 1592 | state.logprobs_processor.logprobs = None |
| 1593 | state.logprobs_processor.cumulative_logprob = None |
| 1594 | state.output_kind = output_kind |
| 1595 | state.request_index = 0 |
| 1596 | state.sampling_mask_chunks = [] |
| 1597 | state.routed_experts_chunks = [] |
| 1598 | state.spec_decode_metrics = None |
| 1599 | |
| 1600 | supports = [[10, 11], [20], [30, 31]] |
| 1601 | masks = [] |
| 1602 | for position, support in enumerate(supports): |
| 1603 | state.sampling_mask_chunks.append(SamplingMaskLists(np.asarray(support))) |
| 1604 | finish = FinishReason.LENGTH if position == len(supports) - 1 else None |
| 1605 | mask = state._new_completion_output([support[0]], finish, None).sampling_mask |
| 1606 | masks.append(mask.token_ids if mask is not None else None) |
| 1607 | |
| 1608 | if output_kind == RequestOutputKind.DELTA: |
| 1609 | assert masks == [[support] for support in supports] |
| 1610 | else: |
| 1611 | assert masks == [None, None, supports] |
| 1612 | |
| 1613 | |
| 1614 | def test_request_output_add_merges_delta_sampling_masks(): |
| 1615 | def delta(token_ids, sampling_mask): |
| 1616 | completion = CompletionOutput( |
| 1617 | index=0, |
| 1618 | text="", |
| 1619 | token_ids=token_ids, |
| 1620 | cumulative_logprob=None, |
| 1621 | logprobs=None, |
| 1622 | sampling_mask=sampling_mask, |
| 1623 | ) |
| 1624 | return RequestOutput( |
| 1625 | request_id="request", |
| 1626 | prompt=None, |
| 1627 | prompt_token_ids=[1], |
| 1628 | prompt_logprobs=None, |
| 1629 | outputs=[completion], |
| 1630 | finished=False, |
| 1631 | ) |
| 1632 | |
| 1633 | merged = delta([], None) |
| 1634 | merged.add(delta([10], SamplingMask([[10, 11]])), aggregate=True) |
| 1635 | merged.add(delta([20], SamplingMask([[20]])), aggregate=True) |
| 1636 | |
| 1637 | assert merged.outputs[0].token_ids == [10, 20] |
| 1638 | assert merged.outputs[0].sampling_mask.token_ids == [[10, 11], [20]] |
| 1639 | |
| 1640 | |
| 1641 | @pytest.mark.skip_global_cleanup |
| 1642 | def test_pooling_cache_miss_is_a_request_error_and_batch_continues(): |
| 1643 | output_processor = OutputProcessor(tokenizer=None, log_stats=False) |
| 1644 | requests = [ |
| 1645 | EngineCoreRequest( |
| 1646 | request_id=f"request-{idx}-int", |
| 1647 | external_req_id=f"request-{idx}", |
| 1648 | prompt_token_ids=[1, 2], |
| 1649 | mm_features=None, |
| 1650 | arrival_time=0, |
| 1651 | lora_request=None, |
| 1652 | cache_salt=None, |
| 1653 | data_parallel_rank=None, |
| 1654 | sampling_params=None, |
| 1655 | pooling_params=PoolingParams(task="embed"), |
| 1656 | ) |
| 1657 | for idx in range(2) |
| 1658 | ] |
| 1659 | for request in requests: |
| 1660 | output_processor.add_request(request, prompt=None) |
| 1661 | |
| 1662 | processed = output_processor.process_outputs( |
| 1663 | [ |
| 1664 | EngineCoreOutput( |
| 1665 | request_id=requests[0].request_id, |
| 1666 | new_token_ids=[], |
| 1667 | finish_reason=FinishReason.ERROR, |
| 1668 | mm_cache_miss_hashes=["missing-hash"], |
| 1669 | ), |
| 1670 | EngineCoreOutput( |
| 1671 | request_id=requests[1].request_id, |
| 1672 | new_token_ids=[], |
| 1673 | pooling_output=torch.tensor([1.0, 2.0]), |
| 1674 | finish_reason=FinishReason.STOP, |
| 1675 | ), |
| 1676 | ] |
| 1677 | ) |
| 1678 | |
| 1679 | assert len(processed.request_outputs) == 2 |
| 1680 | failed, succeeded = processed.request_outputs |
| 1681 | assert isinstance(failed, PoolingRequestOutput) |
| 1682 | assert failed.error is not None |
| 1683 | assert failed.error.code == "multimodal_cache_miss" |
| 1684 | assert failed.error.retryable |
| 1685 | assert isinstance(succeeded, PoolingRequestOutput) |
| 1686 | assert succeeded.error is None |
| 1687 | assert torch.equal(succeeded.outputs.data, torch.tensor([1.0, 2.0])) |
| 1688 | |
| 1689 | |
| 1690 | @pytest.mark.skip_global_cleanup |
| 1691 | def test_async_pooling_cache_miss_does_not_fail_output_processor(): |
| 1692 | output_processor = OutputProcessor(tokenizer=None, log_stats=False) |
| 1693 | request = EngineCoreRequest( |
| 1694 | request_id="request-int", |
| 1695 | external_req_id="request", |
| 1696 | prompt_token_ids=[1, 2], |
| 1697 | mm_features=None, |
| 1698 | arrival_time=0, |
| 1699 | lora_request=None, |
| 1700 | cache_salt=None, |
| 1701 | data_parallel_rank=None, |
| 1702 | sampling_params=None, |
| 1703 | pooling_params=PoolingParams(task="embed"), |
| 1704 | ) |
| 1705 | queue = RequestOutputCollector( |
| 1706 | output_kind=RequestOutputKind.FINAL_ONLY, |
| 1707 | request_id=request.request_id, |
| 1708 | ) |
| 1709 | output_processor.add_request(request, prompt=None, queue=queue) |
| 1710 | |
| 1711 | processed = output_processor.process_outputs( |
| 1712 | [ |
| 1713 | EngineCoreOutput( |
| 1714 | request_id=request.request_id, |
| 1715 | new_token_ids=[], |
| 1716 | finish_reason=FinishReason.ERROR, |
| 1717 | mm_cache_miss_hashes=["missing-hash"], |
| 1718 | ) |
| 1719 | ] |
| 1720 | ) |
| 1721 | |
| 1722 | assert processed.request_outputs == [] |
| 1723 | output = queue.get_nowait() |
| 1724 | assert isinstance(output, PoolingRequestOutput) |
| 1725 | assert output.error is not None |
| 1726 | assert output.error.retryable |
| 1727 | assert not output_processor.has_unfinished_requests() |