(
self,
model_output: paddle.Tensor,
p_done_idxs: List[int],
token_num_event,
model_forward_batch: Optional[List[Request]] = None,
num_running_requests: int = None,
)
| 2354 | return model_output, p_done_idxs, token_num_event |
| 2355 | |
| 2356 | def _postprocess( |
| 2357 | self, |
| 2358 | model_output: paddle.Tensor, |
| 2359 | p_done_idxs: List[int], |
| 2360 | token_num_event, |
| 2361 | model_forward_batch: Optional[List[Request]] = None, |
| 2362 | num_running_requests: int = None, |
| 2363 | ) -> None: |
| 2364 | |
| 2365 | # NOTE(wufeisheng): If `not_need_stop`` is False, it means the current worker is in an idle state. |
| 2366 | # This logic is not used in TP (Tensor Parallelism) mode. However, in EP (Expert Parallelism) mode, |
| 2367 | # Then there is data on other runner, the current runner is required to execute part of the model. |
| 2368 | # But not need to run the below code. |
| 2369 | if not self.not_need_stop(): |
| 2370 | return None, None, None, -1 |
| 2371 | |
| 2372 | if self.use_cudagraph: |
| 2373 | model_output = model_output[: self.real_token_num] |
| 2374 | |
| 2375 | # NOTE(sunxin): |
| 2376 | # token_num_event synchronizes the async DtoH copies of seq_lens_this_time_cpu and is_block_step_cpu, |
| 2377 | # ensuring that the token count for the current batch is ready to be computed and reused in the subsequent batch. |
| 2378 | token_num_event.synchronize() |
| 2379 | if (not self.enable_overlap_schedule) or self.exist_prefill(): |
| 2380 | token_num = -1 |
| 2381 | else: |
| 2382 | token_num = ( |
| 2383 | self.share_inputs["seq_lens_this_time_cpu"].numpy().sum().item() |
| 2384 | + self.share_inputs["is_block_step_cpu"].numpy().sum().item() |
| 2385 | ) |
| 2386 | |
| 2387 | prompt_logprobs_list = self._get_prompt_logprobs_list(model_output) |
| 2388 | if self.is_pooling_model: |
| 2389 | pooler_output = self._pool(model_output, num_running_requests) |
| 2390 | |
| 2391 | model_output_data = ModelOutputData( |
| 2392 | next_tokens=self.share_inputs["next_tokens"], |
| 2393 | stop_flags=self.share_inputs["stop_flags"], |
| 2394 | step_idx=self.share_inputs["step_idx"], |
| 2395 | max_dec_len=self.share_inputs["max_dec_len"], |
| 2396 | pre_ids=self.share_inputs["pre_ids"], |
| 2397 | seq_lens_this_time=self.share_inputs["seq_lens_this_time"], |
| 2398 | eos_token_id=self.share_inputs["eos_token_id"], |
| 2399 | not_need_stop=self.share_inputs["not_need_stop"], |
| 2400 | not_need_stop_device=self.share_inputs["not_need_stop_device"], |
| 2401 | input_ids=self.share_inputs["input_ids"], |
| 2402 | seq_lens_encoder=self.share_inputs["seq_lens_encoder"], |
| 2403 | seq_lens_decoder=self.share_inputs["seq_lens_decoder"], |
| 2404 | is_block_step=self.share_inputs["is_block_step"], |
| 2405 | full_hidden_states=model_output, |
| 2406 | msg_queue_id=self.parallel_config.msg_queue_id, |
| 2407 | mp_rank=self.parallel_config.tensor_parallel_rank, |
| 2408 | use_ep=self.parallel_config.use_ep, |
| 2409 | draft_tokens=(self.share_inputs["draft_tokens"] if self.speculative_decoding else None), |
| 2410 | actual_draft_token_num=( |
| 2411 | self.share_inputs["actual_draft_token_num"] if self.speculative_decoding else None |
| 2412 | ), |
| 2413 | accept_tokens=(self.share_inputs["accept_tokens"] if self.speculative_decoding else None), |
no test coverage detected