Method _build_sample_logprobs

fastdeploy/entrypoints/llm.py:396–437

Constructs a list of dictionaries mapping token IDs to Logprob objects, based on sliced LogprobsLists data (excluding the sampled token at index 0). Args: logprobs_lists (LogprobsLists): Contains top-k token IDs, logprobs, and sampled ranks. max_num

(self, logprobs_lists: LogprobsLists, topk_logprobs: int)

        return self.llm_engine.data_processor.process_logprob_response([token_id], clean_up_tokenization_spaces=False)

    def _build_sample_logprobs(self, logprobs_lists: LogprobsLists, topk_logprobs: int) -> list[dict[int, Logprob]]:
        """
        Constructs a list of dictionaries mapping token IDs to Logprob objects,
        based on sliced LogprobsLists data (excluding the sampled token at index 0).

        Args:
            logprobs_lists (LogprobsLists): Contains top-k token IDs, logprobs, and sampled ranks.
            max_num (int): Maximum number of top logprobs to include (excluding sampled token at index 0).

        Returns:
            list[dict[int, Logprob]]: One dict per request, mapping token ID to Logprob.
        """
        try:
            llm_logger.info(f"filter logprobs, topk_logprobs: {topk_logprobs}")
            if not logprobs_lists.logprob_token_ids:
                llm_logger.warning("Empty logprob_token_ids in LogprobsLists")
                return None

            available_topk = len(logprobs_lists.logprob_token_ids[0])
            effective_topk_logprobs = min(topk_logprobs, available_topk)

            if effective_topk_logprobs < 0:
                llm_logger.warning(
                    f"Invalid effective_topk_logprobs={effective_topk_logprobs}, "
                    f"available_topk={available_topk}, topk_logprobs={topk_logprobs}; returning empty result."
                )
                return None

            # sliced 0 ~ effective_topk_logprobs+1
            sliced_logprobs_lists = logprobs_lists.slice_columns(0, effective_topk_logprobs + 1)
            result = []
            for token_ids, logprobs in zip(sliced_logprobs_lists.logprob_token_ids, sliced_logprobs_lists.logprobs):

                logprob_dict = {
                    token_id: Logprob(logprob=logprob, rank=i + 1, decoded_token=self._decode_token(token_id))
                    for i, (token_id, logprob) in enumerate(zip(token_ids, logprobs))
                }
                result.append(logprob_dict)
            return result

        except Exception as e:
            llm_logger.error(f"Error building sample logprobs from LogprobsLists: {e}, {str(traceback.format_exc())}")

    def _build_prompt_logprobs(
        self,