llm.py

"""
# Copyright (c) 2025  PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""

from __future__ import annotations

import itertools
import logging
import threading
import time
import traceback
import uuid
from collections.abc import Iterable
from typing import Any, Optional, Union

from pydantic import ValidationError
from tqdm import tqdm

from fastdeploy.engine.args_utils import EngineArgs
from fastdeploy.engine.engine import LLMEngine
from fastdeploy.engine.request import RequestOutput
from fastdeploy.engine.sampling_params import SamplingParams
from fastdeploy.entrypoints.chat_utils import load_chat_template
from fastdeploy.entrypoints.openai.protocol import ChatCompletionToolsParam
from fastdeploy.entrypoints.openai.tool_parsers import ToolParserManager
from fastdeploy.input.utils import validate_model_path
from fastdeploy.logger.request_logger import log_request_error
from fastdeploy.utils import (
    deprecated_kwargs_warning,
    llm_logger,
    retrive_model_from_server,
)
from fastdeploy.worker.output import (
    Logprob,
    LogprobsLists,
    LogprobsTensors,
    PromptLogprobs,
)

root_logger = logging.getLogger()
for handler in root_logger.handlers[:]:
    if isinstance(handler, logging.StreamHandler):
        root_logger.removeHandler(handler)

NONES = itertools.repeat(None)


class LLM:
    """
    Initializes a Language Model instance.

    Args:
        model (str):
            The name of the language model to use. Supported models are listed in
            `LLMEngine.SUPPORTED_MODELS`.
        tokenizer (Optional[str], optional):
            The name of the tokenizer to use. Defaults to None. If not specified, the
            default tokenizer for the selected model will be used.
        gpu_memory_utilization: The ratio (between 0 and 1) of GPU memory to
            reserve for the model weights, activations, and KV cache. Higher
            values will increase the KV cache size and thus improve the model's
            throughput. However, if the value is too high, it may cause out-of-
            memory (OOM) errors.
        **kwargs (optional):
            Additional keyword arguments to pass to the `EngineArgs` constructor. See
            `EngineArgs.__init__` for details. Defaults to {}.

    Raises:
        ValueError:
            If `model` is not in `LLMEngine.SUPPORTED_MODELS`.
    """

    def __init__(
        self,
        model: str,
        revision: Optional[str] = "master",
        tokenizer: Optional[str] = None,
        enable_logprob: Optional[bool] = False,
        chat_template: Optional[str] = None,
        **kwargs,
    ):
        deprecated_kwargs_warning(**kwargs)
        validate_model_path(model)

        model = retrive_model_from_server(model, revision)
        tool_parser_plugin = kwargs.get("tool_parser_plugin")
        if tool_parser_plugin:
            ToolParserManager.import_tool_parser(tool_parser_plugin)
        engine_args = EngineArgs(
            model=model,
            tokenizer=tokenizer,
            enable_logprob=enable_logprob,
            **kwargs,
        )

        # Create the Engine
        self.llm_engine = LLMEngine.from_engine_args(engine_args=engine_args)

        self.default_sampling_params = SamplingParams(max_tokens=self.llm_engine.cfg.model_config.max_model_len)

        self.llm_engine.start()

        self.mutex = threading.Lock()
        self.req_output = dict()
        self.master_node_ip = self.llm_engine.cfg.master_ip
        self._receive_output_thread = threading.Thread(target=self._receive_output, daemon=True)
        self._receive_output_thread.start()
        self.chat_template = load_chat_template(chat_template, model)

    def _check_master(self):
        """
        Check if the current node is the master node.
        """
        return self.llm_engine.cfg._check_master()

    def _receive_output(self):
        """
        Receive output from token processor and store them in cache
        """
        while True:
            try:
                results = self.llm_engine._get_generated_result()
                for request_id, contents in results.items():
                    with self.mutex:
                        for result in contents:
                            if request_id not in self.req_output:
                                self.req_output[request_id] = result
                                continue
                            self.req_output[request_id].add(result)
            except Exception as e:
                log_request_error(
                    message="Unexpected error happened: {error}, {traceback}",
                    error=e,
                    traceback=traceback.format_exc(),
                )

    def generate(
        self,
        prompts: Union[
            str,
            list[str],
            list[int],
            list[list[int]],
            dict[str, Any],
            list[dict[str, Any]],
        ],
        sampling_params: Optional[Union[SamplingParams, list[SamplingParams]]] = None,
        use_tqdm: bool = True,
        stream: bool = False,
    ):
        """
        Generate function for the LLM class.

        Args:
            prompts (Union[str, list[str], list[int], list[list[int]], dict[str, Any], list[dict[str, Any]]]):
                The prompt to use for generating the response.
            sampling_params (Optional[Union[SamplingParams, list[SamplingParams]]], optional):
                The sampling parameters to use for generating the response. Defaults to None.
            use_tqdm (bool, optional): Whether to use tqdm for the progress bar. Defaults to True.
            stream (bool, optional): Whether to return a streaming iterator. Defaults to False.

        Returns:
            If stream=False: Union[str, list[str]]: The generated response.
            If stream=True: Iterator: An iterator that yields partial responses as they become available.
        """

        if not self._check_master():
            err_msg = f"Only master node can accept completion request, please send request to master node: {self.master_node_ip}"
            raise ValueError(err_msg)

        if sampling_params is None:
            sampling_params = self.default_sampling_params

        if isinstance(sampling_params, SamplingParams):
            sampling_params_len = 1
        else:
            sampling_params_len = len(sampling_params)

        if isinstance(prompts, str):
            prompts = [prompts]

        if isinstance(prompts, list) and isinstance(prompts[0], int):
            prompts = [prompts]

        if isinstance(prompts, dict):
            if "prompt" not in prompts:
                raise ValueError("prompts must be a input dict")
            prompts = [prompts]
            # sampling_params = None

        if sampling_params_len != 1 and len(prompts) != sampling_params_len:
            raise ValueError("prompts and sampling_params must be the same length.")

        req_ids = self._add_request(prompts=prompts, sampling_params=sampling_params)

        topk_logprobs = sampling_params[0].logprobs if sampling_params_len > 1 else sampling_params.logprobs
        num_prompt_logprobs = (
            sampling_params[0].prompt_logprobs if sampling_params_len > 1 else sampling_params.prompt_logprobs
        )

        # get output
        if stream:
            return self._run_engine_stream(req_ids, prompts, use_tqdm=use_tqdm, topk_logprobs=topk_logprobs)
        else:
            outputs = self._run_engine(
                req_ids, use_tqdm=use_tqdm, topk_logprobs=topk_logprobs, num_prompt_logprobs=num_prompt_logprobs
            )
            for i in range(len(outputs)):
                outputs[i].prompt = prompts[i]
            return outputs

    def chat(
        self,
        messages: Union[list[Any], list[list[Any]]],
        sampling_params: Optional[Union[SamplingParams, list[SamplingParams]]] = None,
        use_tqdm: bool = True,
        chat_template_kwargs: Optional[dict[str, Any]] = None,
        chat_template: Optional[str] = None,
        tools: Optional[Union[ChatCompletionToolsParam, list[ChatCompletionToolsParam]]] = None,
        stream: bool = False,
    ):
        """
        Args:
            messages (Union[list[ChatCompletionMessageParam], list[list[ChatCompletionMessageParam]]]):
                Single conversation or a list of conversations.
            sampling_params (Optional[Union[SamplingParams, list[SamplingParams]]], optional):
                The sampling parameters to use for generating the response. Defaults to None.
            use_tqdm (bool, optional): Whether to use tqdm for the progress bar. Defaults to True.
            chat_template_kwargs(Optional[dict[str,Any]]): Additional kwargs to pass to the chat
                template.
            stream (bool, optional): Whether to return a streaming iterator. Defaults to False.

        Returns:
            If stream=False: Union[str, list[str]]: The generated response.
            If stream=True: Iterator: An iterator that yields partial responses as they become available.
        """

        if not self._check_master():
            err_msg = f"Only master node can accept completion request, please send request to master node: {self.master_node_ip}"
            raise ValueError(err_msg)

        if sampling_params is None:
            sampling_params = self.default_sampling_params

        if isinstance(sampling_params, SamplingParams):
            sampling_params_len = 1
        else:
            sampling_params_len = len(sampling_params)

        if isinstance(messages, list) and isinstance(messages[0], dict):
            messages = [messages]

        if sampling_params_len != 1 and len(messages) != sampling_params_len:
            raise ValueError("messages and sampling_params must be the same length.")

        if chat_template is None:
            chat_template = self.chat_template

        validated_tools = None
        if tools is not None:
            try:
                validated_tools = self._validate_tools(tools)
            except ValueError as e:
                raise RuntimeError(f"Failed to validate 'tools' parameter in chat method: {e}") from e

        req_ids = self._add_request(
            prompts=messages,
            sampling_params=sampling_params,
            chat_template_kwargs=chat_template_kwargs,
            chat_template=chat_template,
            tools=validated_tools,
        )

        topk_logprobs = sampling_params[0].logprobs if sampling_params_len > 1 else sampling_params.logprobs

        # get output
        if stream:
            return self._run_engine_stream(
                req_ids,
                messages,
                use_tqdm=use_tqdm,
                topk_logprobs=topk_logprobs,
                chat_template_kwargs=chat_template_kwargs,
            )
        else:
            outputs = self._run_engine(req_ids, use_tqdm=use_tqdm, topk_logprobs=topk_logprobs)
            return outputs

    def _add_request(
        self,
        prompts,
        sampling_params,
        **kwargs,
    ):
        """
            添加一个请求到 LLM Engine，并返回该请求的 ID。
        如果请求已经存在于 LLM Engine 中，则不会重复添加。

        Args:
            prompts (str): 需要处理的文本内容，类型为字符串。

        Returns:
            None: 无返回值，直接修改 LLM Engine 的状态。
        """
        if prompts is None:
            raise ValueError("prompts and prompt_ids cannot be both None.")

        prompts_len = len(prompts)
        req_ids = []
        for i in range(prompts_len):
            request_id = str(uuid.uuid4())
            if isinstance(prompts[i], str):
                tasks = {
                    "prompt": prompts[i],
                    "request_id": request_id,
                }
            elif isinstance(prompts[i], list) and len(prompts[i]) == 0:
                raise ValueError(
                    f"prompts[{i}] is an empty list. Expected a non-empty list of int (prompt_token_ids) "
                    "or a non-empty list of dict (messages)."
                )
            elif isinstance(prompts[i], list) and isinstance(prompts[i][0], int):
                tasks = {
                    "prompt_token_ids": prompts[i],
                    "request_id": request_id,
                }
            elif isinstance(prompts[i], list) and isinstance(prompts[i][0], dict):
                tasks = {
                    "messages": prompts[i],
                    "request_id": request_id,
                }
            elif isinstance(prompts[i], dict):
                tasks = prompts[i]
                tasks["request_id"] = request_id
            else:
                raise TypeError(
                    f"Invalid type for 'prompt': {type(prompts[i])}, expected one of ['str', 'list', 'dict']."
                )
            req_ids.append(request_id)
            if isinstance(sampling_params, list):
                current_sampling_params = sampling_params[i]
            else:
                current_sampling_params = sampling_params

            ori_vocab_size = (
                len(self.llm_engine.data_processor.tokenizer.sp_model)
                if hasattr(self.llm_engine.data_processor.tokenizer, "sp_model")
                else len(self.llm_engine.data_processor.tokenizer.vocab)
            )
            max_logprobs = self.llm_engine.cfg.model_config.max_logprobs
            if max_logprobs == -1:
                max_logprobs = ori_vocab_size
            if max_logprobs < -1:
                raise ValueError(f"max_logprobs ({max_logprobs}) can't be less than -1.")
            if max_logprobs > ori_vocab_size:
                raise ValueError(f"max_logprobs ({max_logprobs}) exceeds vocabulary size ({ori_vocab_size}).")

            if current_sampling_params.logprobs is not None:
                num_logprobs = current_sampling_params.logprobs
                if not self.llm_engine.cfg.model_config.enable_logprob:
                    raise ValueError(
                        "logprobs is only supported if `enable_logprob` is set to true in startup config."
                    )
                if num_logprobs == -1 and ori_vocab_size > max_logprobs:
                    raise ValueError(
                        f"Number of logprobs(-1) requested ({ori_vocab_size}) exceeds maximum allowed value ({max_logprobs})."
                    )
                if num_logprobs > max_logprobs:
                    raise ValueError(
                        f"Number of logprobs requested ({num_logprobs}) exceeds maximum allowed value ({max_logprobs})."
                    )
            if current_sampling_params.prompt_logprobs is not None:
                if not self.llm_engine.cfg.model_config.enable_logprob:
                    raise ValueError(
                        "prompt_logprobs is only supported if `enable_logprob` is set to true in startup config."
                    )
                if self.llm_engine.cfg.cache_config.enable_prefix_caching:
                    raise ValueError("prompt_logprobs is not supported with prefix caching enabled.")
                if kwargs.get("stream"):
                    raise ValueError("prompt_logprobs is not supported with streaming.")
                num_prompt_logprobs = current_sampling_params.prompt_logprobs
                if num_prompt_logprobs == -1 and ori_vocab_size > max_logprobs:
                    raise ValueError(
                        f"Number of prompt_logprobs(-1) requested ({ori_vocab_size}) exceeds maximum allowed value ({max_logprobs})."
                    )
                if num_prompt_logprobs > max_logprobs:
                    raise ValueError(
                        f"Number of logprobs requested ({num_prompt_logprobs}) exceeds maximum allowed value ({max_logprobs})."
                    )
            if current_sampling_params.guided_decoding is not None:
                guided_decoding_dict = current_sampling_params.guided_decoding.to_dict()
                tasks.update(guided_decoding_dict)
            if kwargs.get("tools") is not None:
                tasks["tools"] = kwargs.get("tools")
            self.llm_engine.add_requests(tasks, current_sampling_params, **kwargs)
        return req_ids

    def _decode_token(self, token_id: int) -> str:
        """Decodes a single token ID into its string representation."""
        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:
            log_request_error(
                message="Error building sample logprobs from LogprobsLists: {error}, {traceback}",
                error=e,
                traceback=traceback.format_exc(),
            )

    def _build_prompt_logprobs(
        self,
        prompt_logprobs_tensors: LogprobsTensors,
        num_prompt_logprobs: int,
    ):
        """Update with prompt logprobs from worker.
        Args:
          prompt_logprobs_tensors: tuple containing the prompt logprobs
                                   tensors.
        """

        token_ids, logprobs, ranks = prompt_logprobs_tensors

        # Detokenize non-incrementally.
        # Output is flat: [num_tok, num_lps] -> [num_tok * num_lps]
        decoded_tokens = [self._decode_token(token_id) for token_id in token_ids.flatten().tolist()]

        # Recover shapes.
        num_prompt_tokens, num_logprobs = logprobs.shape

        # Pythonize the paddle tensors.
        prompt_token_ranks = ranks.tolist()
        prompt_logprobs = logprobs.tolist()
        token_ids = token_ids.tolist()
        result: Optional[PromptLogprobs] = [None]
        # Make Logprob for each position.
        for pos in range(num_prompt_tokens):
            # Handle flattening.
            offset = pos * num_logprobs
            offset_end = offset + num_logprobs
            decoded_tokens_for_pos = NONES if decoded_tokens is None else decoded_tokens[offset:offset_end]

            # Update with the Logprob dictionary for this pos.
            result.append(
                self._make_logprob_dict(
                    prompt_logprobs[pos],
                    token_ids[pos],
                    decoded_tokens_for_pos,
                    prompt_token_ranks[pos],
                    num_prompt_logprobs,
                )
            )
        return result

    @staticmethod
    def _make_logprob_dict(
        logprobs: list[float],
        logprob_token_ids: list[int],
        decoded_tokens: Iterable[str | None],
        rank: int,
        num_logprobs: int,
    ) -> dict[int, Logprob]:
        """Make a Logprob dictionary for a position.
        Args:
          logprobs: list of log probabilities
          logprob_token_ids: list of top token ids
          decoded_tokens: list of decoded top tokens
          rank: rank of the sampled token
          num_logprobs: number of logprobs requested
            by the user (in addition to sampled logprob)
        Returns:
          dict[token id, Logprob]
        """
        if num_logprobs == -1:
            num_logprobs = len(logprobs)
        # We do not need a special case for the sampled token
        # being in the topk, since inserting duplicated data
        # into a dictionary twice is the same as doing it once.
        topk_ranks = range(1, num_logprobs + 1)
        ranks = itertools.chain((rank,), topk_ranks)

        return {
            token_id: Logprob(
                logprob=logprob,
                rank=rank,
                decoded_token=token,
            )
            for token_id, logprob, rank, token in zip(logprob_token_ids, logprobs, ranks, decoded_tokens)
        }

    def _run_engine(
        self,
        req_ids: list[str],
        use_tqdm: bool,
        topk_logprobs: Optional[int] = None,
        num_prompt_logprobs: Optional[int] = None,
    ):
        """
            运行引擎，并返回结果列表。

        Args:
            use_tqdm (bool, optional): 是否使用tqdm进度条，默认为False。

        Returns:
            list[Dict[str, Any]]: 包含每个请求的结果字典的列表，字典中包含以下键值对：
                    - "text": str, 生成的文本；
                    - "score": float, 得分（可选）。

        Raises:
            无。
        """
        # Initialize tqdm.

        if use_tqdm:
            num_requests = len(req_ids)
            pbar = tqdm(
                total=num_requests,
                desc="Processed prompts",
                dynamic_ncols=True,
                postfix=(f"est. speed input: {0:.2f} toks/s, " f"output: {0:.2f} toks/s"),
            )

        output = [None] * num_requests
        req_ids = [(pos, req_id) for pos, req_id in enumerate(req_ids)]
        while num_requests:
            finished = []
            for i, (pos, req_id) in enumerate(req_ids):
                with self.mutex:
                    if req_id not in self.req_output:
                        time.sleep(0.01)
                        continue

                    if not self.req_output[req_id].finished:
                        time.sleep(0.01)
                        continue

                    result = self.req_output.pop(req_id)
                    result_dict = result.to_dict()
                    result_dict = self.llm_engine.data_processor.process_response_dict(
                        result_dict, stream=False, include_stop_str_in_output=False, direct_decode=True
                    )
                    result = RequestOutput.from_dict(result_dict)

                    # filter logprobs
                    if result.outputs.top_logprobs is not None and topk_logprobs is not None:
                        if topk_logprobs == -1:
                            topk_logprobs = self.llm_engine.cfg.model_config.ori_vocab_size
                        result.outputs.logprobs = self._build_sample_logprobs(
                            result.outputs.top_logprobs, topk_logprobs
                        )
                    if result.prompt_logprobs is not None and num_prompt_logprobs is not None:
                        if num_prompt_logprobs == -1:
                            num_prompt_logprobs = self.llm_engine.cfg.model_config.ori_vocab_size
                        result.prompt_logprobs = self._build_prompt_logprobs(
                            result.prompt_logprobs, num_prompt_logprobs
                        )

                    output[pos] = result
                    finished.append(i)

                    llm_logger.debug(f"Request id: {req_id} has been completed.")

                    if use_tqdm:
                        pbar.update(1)

            num_requests -= len(finished)
            for i in reversed(finished):
                req_ids.pop(i)

        if use_tqdm:
            pbar.close()
        return output

    def _run_engine_stream(
        self,
        req_ids: list[str],
        prompts,
        use_tqdm: bool,
        topk_logprobs: Optional[int] = None,
        chat_template_kwargs: Optional[dict[str, Any]] = None,
    ):
        """
        运行引擎并返回流式响应的迭代器。

        Args:
            req_ids (list[str]): 请求ID列表
            prompts: 原始提示词列表，用于设置到输出中
            use_tqdm (bool, optional): 是否使用tqdm进度条
            topk_logprobs (Optional[int]): 返回的top-k logprobs数量

        Yields:
            list[RequestOutput]: 包含增量更新的部分响应列表
        """
        # Initialize tqdm
        if use_tqdm:
            num_requests = len(req_ids)
            pbar = tqdm(
                total=num_requests,
                desc="Processed prompts",
                dynamic_ncols=True,
                postfix=(f"est. speed input: {0:.2f} toks/s, " f"output: {0:.2f} toks/s"),
            )

        num_requests = len(req_ids)
        original_num_requests = len(req_ids)  # Keep track of original count
        output = [None] * original_num_requests
        req_ids_with_pos = [(pos, req_id) for pos, req_id in enumerate(req_ids)]

        # Track previous token counts for each request to identify new tokens
        previous_token_counts = {req_id: 0 for req_id in req_ids}

        while num_requests > 0:
            has_new_tokens = False
            finished = []

            for i, (pos, req_id) in enumerate(req_ids_with_pos):
                with self.mutex:
                    if req_id not in self.req_output:
                        continue

                    current_result = self.req_output[req_id]
                    current_token_count = (
                        len(current_result.outputs.token_ids) if current_result.outputs.token_ids else 0
                    )
                    previous_count = previous_token_counts[req_id]

                    # Check if there are new tokens since last yield
                    if current_token_count > previous_count:
                        has_new_tokens = True
                        # Create incremental output with only new tokens
                        incremental_result = self._create_incremental_result(
                            current_result, previous_count, pos, prompts, chat_template_kwargs
                        )

                        # Apply logprobs filtering to the incremental result if needed
                        if incremental_result.outputs.top_logprobs and topk_logprobs:
                            incremental_result.outputs.logprobs = self._build_sample_logprobs(
                                incremental_result.outputs.top_logprobs, topk_logprobs
                            )

                        output[pos] = incremental_result
                        previous_token_counts[req_id] = current_token_count

                    # Check if request is finished
                    if current_result.finished:
                        finished.append(i)

                        # For streaming, when a request is finished, we should NOT output anything
                        self.req_output.pop(req_id)

                        llm_logger.debug(f"Request id: {req_id} has been completed.")

                        if use_tqdm:
                            pbar.update(1)

            # Yield updates if there are new tokens
            if has_new_tokens or finished:
                # yield [result for result in output if result is not None]
                # Create a complete output array with proper indexing
                complete_output = [None] * original_num_requests  # Use original length
                for i, (pos, _) in enumerate(req_ids_with_pos):
                    if output[pos] is not None:
                        complete_output[pos] = output[pos]
                yield complete_output
                # Clear output for next iteration
                output = [None] * original_num_requests

            # Remove finished requests
            num_requests -= len(finished)
            for i in reversed(finished):
                req_ids_with_pos.pop(i)

            if num_requests > 0:
                time.sleep(0.01)

        if use_tqdm:
            pbar.close()

    def _create_incremental_result(
        self, current_result, previous_count, pos, prompts, chat_template_kwargs: Optional[dict[str, Any]] = None
    ):
        """
        创建包含增量token的结果对象

        Args:
            current_result: 当前的RequestOutput对象
            previous_count: 之前已处理的token数量
            pos: 在prompts列表中的位置
            prompts: 原始提示词列表
            chat_template_kwargs: 聊天模板参数，包含enable_thinking等配置

        Returns:
            RequestOutput: 包含增量更新的结果对象
        """
        # Create a copy of current result for incremental update
        from copy import deepcopy

        incremental_result = deepcopy(current_result)

        # Extract only new tokens
        if current_result.outputs.token_ids and len(current_result.outputs.token_ids) > previous_count:
            new_token_ids = current_result.outputs.token_ids[previous_count:]
            incremental_result.outputs.token_ids = new_token_ids

            # Get enable_thinking from chat_template_kwargs, default to False
            enable_thinking = False
            if chat_template_kwargs:
                enable_thinking = chat_template_kwargs.get("enable_thinking", False)

            # Construct response_dict format and call process_response_dict_streaming
            response_dict = {
                "request_id": current_result.request_id,
                "finished": current_result.finished,
                "outputs": {
                    "token_ids": new_token_ids,
                },
            }

            processed_response = self.llm_engine.data_processor.process_response_dict_streaming(
                response_dict, stream=True, enable_thinking=enable_thinking, include_stop_str_in_output=False
            )

            # Extract incremental text
            incremental_result.outputs.text = processed_response["outputs"]["text"]

        # Set the prompt
        if isinstance(prompts, list):
            incremental_result.prompt = prompts[pos]
        else:
            incremental_result.prompt = prompts

        return incremental_result

    def _validate_tools(self, raw_tools: Any) -> Optional[list[dict]]:
        """
        Validate the format of the `tools` parameter for chat requests.
        Valid inputs are accepted and standardized, while invalid inputs raise ValueError.
        Empty dict/list will be returned as None.

        Args:
            raw_tools: Raw `tools` parameter obtained from kwargs (can be any type)

        Returns:
            Optional[List[Dict[str, Any]]]: Standardized list of valid tool dictionaries if validation passes;
            None if `raw_tools` is None or empty (empty dict/list).

        Raises:
            ValueError: Raised when input type is invalid or format does not meet standards.
        """
        if raw_tools is None:
            return None
        if isinstance(raw_tools, ChatCompletionToolsParam):
            return [raw_tools]
        if isinstance(raw_tools, list) and all(isinstance(t, ChatCompletionToolsParam) for t in raw_tools):
            if not raw_tools:
                return None
            else:
                return raw_tools

        if not isinstance(raw_tools, dict) and not isinstance(raw_tools, list):
            raise ValueError(
                f"Invalid tools top-level type! Expected None, dict (single tool) or list (multiple tools), "
                f"but got type '{type(raw_tools).__name__}' (value: {raw_tools})."
            )
        tools_list: list[dict[str, Any]] = [raw_tools] if isinstance(raw_tools, dict) else raw_tools

        if not tools_list:
            return None

        validated_tools = []
        for idx, tool in enumerate(tools_list):
            if not isinstance(tool, dict):
                raise ValueError(
                    f"Invalid element type in tools list! At index {idx}, "
                    f"expected dict (tool definition), but got type '{type(tool).__name__}' (value: {tool})."
                )

            try:
                validated_tool_obj = ChatCompletionToolsParam.model_validate(tool)
                validated_tools.append(validated_tool_obj.model_dump())
            except ValidationError as e:
                raise ValueError(
                    f"Invalid tool format at index {idx} in tools list! " f"Tool content: {tool}\nError details: {e}"
                ) from e

        return validated_tools


if __name__ == "__main__":
    # llm = LLM(model="llama_model")
    # output = llm.generate(prompts="who are you？", use_tqdm=True)
    # print(output)

    llm = LLM(
        model="/opt/baidu/paddle_internal/FastDeploy/Qwen2.5-7B",
        tensor_parallel_size=2,
    )
    sampling_params = SamplingParams(temperature=0.1, max_tokens=30)
    output = llm.generate(prompts="who are you？", use_tqdm=True, sampling_params=sampling_params)
    print(output)

    output = llm.generate(
        prompts=["who are you？", "what can you do？"],
        sampling_params=SamplingParams(temperature=1, max_tokens=50),
        use_tqdm=True,
    )
    print(output)

    output = llm.generate(
        prompts=["who are you？", "I miss you"],
        sampling_params=[
            SamplingParams(temperature=1, max_tokens=50),
            SamplingParams(temperature=1, max_tokens=20),
        ],
        use_tqdm=True,
    )
    print(output)