support stop_reqs

Author: zoooo0820

docs/usage/environment_variables.md

@@ -32,6 +32,10 @@ environment_variables: dict[str, Callable[[], Any]] = {
     "FD_STOP_SEQS_MAX_LEN":
     lambda: os.getenv("FD_STOP_SEQS_MAX_LEN", "8"),
 
+    # Whether to use stop sequences (0 or 1)
+    "FD_USE_STOP_SEQ":
+    lambda: os.getenv("FD_USE_STOP_SEQ", 0),
+
     # GPU devices to use (comma-separated string, e.g. 0,1,2)
     "CUDA_VISIBLE_DEVICES":
     lambda: os.getenv("CUDA_VISIBLE_DEVICES", None),
@@ -67,6 +71,6 @@ environment_variables: dict[str, Callable[[], Any]] = {
     # Switch from standalone PD to centralized inference (0 or 1)
     "FD_PD_CHANGEABLE":
     lambda: os.getenv("FD_PD_CHANGEABLE", "1"),
-  
+
 }
-```
+```

docs/zh/usage/environment_variables.md

@@ -1,5 +1,6 @@
 # FastDeploy 环境变量说明
 FastDeploy 的环境变量保存在了代码库根目录下 fastdeploy/envs.py 文件中，以下是其对应的中文版说明：
+
 ```python
 environment_variables: dict[str, Callable[[], Any]] = {
     # 构建 FastDeploy 时使用的 CUDA 架构版本，这是一个字符串列表，例如[80,90]
@@ -30,6 +31,10 @@ environment_variables: dict[str, Callable[[], Any]] = {
     "FD_STOP_SEQS_MAX_LEN":
     lambda: os.getenv("FD_STOP_SEQS_MAX_LEN", "8"),
 
+    # 是否使用停止序列
+    "FD_USE_STOP_SEQ":
+    lambda: os.getenv("FD_USE_STOP_SEQ", 0),
+
     # 将要使用的GPU设备，这是一个用逗号分隔的字符串，例如 0,1,2
     "CUDA_VISIBLE_DEVICES":
     lambda: os.getenv("CUDA_VISIBLE_DEVICES", None),
@@ -65,6 +70,6 @@ environment_variables: dict[str, Callable[[], Any]] = {
     # 是否从单机 PD 分离转换为集中式推理
     "FD_PD_CHANGEABLE":
     lambda: os.getenv("FD_PD_CHANGEABLE", "1"),
-  
+
 }
-```
+```

fastdeploy/config.py

@@ -22,6 +22,7 @@
 
 from paddleformers.transformers.configuration_utils import PretrainedConfig
 
+from fastdeploy import envs
 from fastdeploy.model_executor.layers.quantization.quant_base import \
     QuantConfigBase
 from fastdeploy.utils import get_logger
@@ -124,6 +125,9 @@ def __init__(
         self.tie_word_embeddings = tie_word_embeddings
         self.is_quantized = is_quantized
 
+        self.max_stop_seqs_num = int(envs.FD_MAX_STOP_SEQS_NUM)
+        self.stop_seqs_max_len = int(envs.FD_STOP_SEQS_MAX_LEN)
+        self.use_stop_seq = int(envs.FD_USE_STOP_SEQ)
 
 @dataclass
 class MoEConfig:

fastdeploy/engine/config.py

@@ -126,6 +126,7 @@ def read_from_env(self):
         """
         self.max_stop_seqs_num = int(envs.FD_MAX_STOP_SEQS_NUM)
         self.stop_seqs_max_len = int(envs.FD_STOP_SEQS_MAX_LEN)
+        self.use_stop_seq = int(envs.FD_USE_STOP_SEQ)
 
         def reset_config_value(key, value):
             if not hasattr(self, key.lower()):

fastdeploy/engine/sampling_params.py

@@ -85,6 +85,7 @@ class SamplingParams:
     seed: Optional[int] = None
     stop: Optional[Union[str, List[str]]] = None
     stop_token_ids: Optional[Union[List[List[int]], List[int]]] = None
+    stop_seqs_len: Optional[int] = None
     max_tokens: Optional[int] = None
     reasoning_max_tokens: Optional[int] = None
     min_tokens: int = 1

fastdeploy/envs.py

@@ -52,6 +52,10 @@
     "FD_STOP_SEQS_MAX_LEN":
     lambda: os.getenv("FD_STOP_SEQS_MAX_LEN", "8"),
 
+    # Whether to use stop sequences (0 or 1)
+    "FD_USE_STOP_SEQ":
+    lambda: os.getenv("FD_USE_STOP_SEQ", "0"),
+
     # GPU devices that will be used. This is a string that
     # splited by comma, such as 0,1,2.
     "CUDA_VISIBLE_DEVICES":

fastdeploy/model_executor/pre_and_post_process.py

@@ -21,11 +21,12 @@
 from fastdeploy.engine.config import SpeculativeConfig
 from fastdeploy.model_executor.ops.gpu import (
     get_padding_offset, save_output, set_stop_value_multi_ends,
-    speculate_clear_accept_nums, speculate_get_output_padding_offset,
-    speculate_get_padding_offset, speculate_get_seq_lens_output,
-    speculate_save_output, speculate_set_value_by_flags_and_idx,
-    speculate_step_paddle, speculate_step_system_cache, speculate_update_v3,
-    step_paddle, step_system_cache, update_inputs, step_reschedule)
+    set_stop_value_multi_seqs, speculate_clear_accept_nums,
+    speculate_get_output_padding_offset, speculate_get_padding_offset,
+    speculate_get_seq_lens_output, speculate_save_output,
+    speculate_set_value_by_flags_and_idx, speculate_step_paddle,
+    speculate_step_system_cache, speculate_update_v3, step_paddle,
+    step_reschedule, step_system_cache, update_inputs)
 from fastdeploy.platforms import current_platform
 from fastdeploy.worker.output import ModelOutputData
 
@@ -105,7 +106,8 @@ def pre_process(
 def post_process_normal(sampled_token_ids: paddle.Tensor,
                         model_output: ModelOutputData,
                         save_each_rank: bool = False,
-                        skip_save_output: bool = False) -> None:
+                        skip_save_output: bool = False,
+                        use_stop_seqs: bool = False) -> None:
     """ Post-processing steps after completing a single token generation. """
     # 1. Set stop value
     paddle.assign(
@@ -122,12 +124,23 @@ def post_process_normal(sampled_token_ids: paddle.Tensor,
         paddle.logical_or(model_output.stop_flags, length_cond),
         model_output.stop_flags,
     )
-    # TODO(gongshaotian): Add use_stop_seqs
-    set_stop_value_multi_ends(sampled_token_ids, model_output.stop_flags,
-                              model_output.seq_lens_this_time,
-                              model_output.eos_token_id,
-                              model_output.next_tokens, False)  # multi ends
 
+    if not use_stop_seqs:
+        set_stop_value_multi_ends(sampled_token_ids, model_output.stop_flags,
+                                model_output.seq_lens_this_time,
+                                model_output.eos_token_id,
+                                model_output.next_tokens, False)  # multi ends
+    else:
+        set_stop_value_multi_seqs(
+            sampled_token_ids,
+            model_output.pre_ids,
+            model_output.step_idx,
+            model_output.stop_flags,
+            model_output.seq_lens_this_time,
+            model_output.stop_token_ids,
+            model_output.stop_seqs_len,
+            model_output.eos_token_id,
+        )
     # 2. Update the input buffer of the model
     with paddle.framework._no_check_dy2st_diff():
         update_inputs(
@@ -197,13 +210,14 @@ def post_process(sampled_token_ids: paddle.Tensor,
                  model_output: ModelOutputData,
                  save_each_rank: bool = False,
                  speculative_decoding: bool = False,
-                 skip_save_output: bool = False) -> None:
+                 skip_save_output: bool = False,
+                 use_stop_seq: bool = False) -> None:
     """ Post-processing steps after completing a single token generation. """
     if speculative_decoding:
         post_process_specualate(model_output, skip_save_output)
     else:
         post_process_normal(sampled_token_ids, model_output, save_each_rank,
-                            skip_save_output)
+                            skip_save_output, use_stop_seq)
 
 
 def step_cuda(
@@ -217,7 +231,7 @@ def step_cuda(
     TODO(gongshaotian): normalization name
     """
 
-    
+
     if speculative_config.method is not None:
         if enable_prefix_caching:
             speculate_step_system_cache(

fastdeploy/worker/gpu_model_runner.py

@@ -280,9 +280,9 @@ def insert_prefill_inputs(self, req_dicts: List[Request]):
                 stop_seqs_num = len(request.get("stop_seqs_len"))
                 for i in range(stop_seqs_num,
                                self.model_config.max_stop_seqs_num):
-                    request.stop_seqs_len.append(0)
+                    request.sampling_params.stop_seqs_len.append(0)
                 self.share_inputs["stop_seqs_len"][:] = np.array(
-                    request.stop_seqs_len, dtype="int32")
+                    request.sampling_params.stop_seqs_len, dtype="int32")
                 self.share_inputs["stop_seqs"][:stop_seqs_num, :len(
                     request.get("stop_token_ids")[0])] = np.array(
                         request.get("stop_token_ids"), dtype="int64")
@@ -505,7 +505,7 @@ def _init_share_inputs(self, max_num_seqs: int):
             self.model_config.stop_seqs_max_len
         ],
                                                      -1,
-                                                     dtype="int32")
+                                                     dtype="int64")
         if self.speculative_decoding:
             max_draft_token_num = self.speculative_config.num_speculative_tokens
             self.share_inputs["input_ids_cpu"] = paddle.full(
@@ -832,7 +832,11 @@ def _dummy_run(self,
                 accept_tokens=self.share_inputs["accept_tokens"]
                 if self.speculative_decoding else None,
                 accept_num=self.share_inputs["accept_num"]
-                if self.speculative_decoding else None)
+                if self.speculative_decoding else None,
+                stop_token_ids=self.share_inputs["stop_seqs"]
+                if self.model_config.use_stop_seq else None,
+                stop_seqs_len=self.share_inputs["stop_seqs_len"]
+                if self.model_config.use_stop_seq else None)
 
             post_process(sampled_token_ids=sampled_token_ids,
                          model_output=model_output_data,
@@ -1065,7 +1069,12 @@ class at the server level, which is too granular for ModelRunner.
             accept_tokens=self.share_inputs["accept_tokens"]
             if self.speculative_decoding else None,
             accept_num=self.share_inputs["accept_num"]
-            if self.speculative_decoding else None)
+            if self.speculative_decoding else None,
+            stop_token_ids=self.share_inputs["stop_seqs"]
+            if self.model_config.use_stop_seq else None,
+            stop_seqs_len=self.share_inputs["stop_seqs_len"]
+            if self.model_config.use_stop_seq else None,
+        )
 
         if self.speculative_config.method in ["mtp"] and \
             self.parallel_config.splitwise_role == "prefill":
@@ -1076,7 +1085,8 @@ class at the server level, which is too granular for ModelRunner.
                      model_output=model_output_data,
                      save_each_rank=self.parallel_config.use_ep,
                      speculative_decoding=self.speculative_decoding,
-                     skip_save_output=skip_save_output)
+                     skip_save_output=skip_save_output,
+                     use_stop_seq=self.model_config.use_stop_seq)
 
         # 6. Speculative decode
         if self.speculative_decoding:

fastdeploy/worker/output.py

@@ -132,6 +132,15 @@ class ModelOutputData:
     """
     accept_num: paddle.Tensor
 
+    """
+        the token ids of stop sequence
+    """
+    stop_token_ids: paddle.Tensor
+
+    """
+        the length of stop sequence
+    """
+    stop_seqs_len: paddle.Tensor
 
 @dataclass
 class ModelRunnerOutput:

fastdeploy/worker/xpu_model_runner.py

@@ -320,9 +320,9 @@ def process_prefill_inputs(self, req_dicts: List[Request]):
                 stop_seqs_num = len(request.get("stop_seqs_len"))
                 for i in range(stop_seqs_num,
                                self.model_config.max_stop_seqs_num):
-                    request.stop_seqs_len.append(0)
+                    request.sampling_params.stop_seqs_len.append(0)
                 self.share_inputs["stop_seqs_len"][:] = np.array(
-                    request.stop_seqs_len, dtype="int32")
+                    request.sampling_params.stop_seqs_len, dtype="int32")
                 self.share_inputs["stop_seqs"][:stop_seqs_num, :len(
                     request.get("stop_token_ids")[0])] = np.array(
                         request.get("stop_token_ids"), dtype="int64")
@@ -719,6 +719,8 @@ class at the server level, which is too granular for ModelRunner.
             actual_draft_token_num=None,
             accept_tokens=None,
             accept_num=None,
+            stop_token_ids=None,
+            stop_seqs_len=None,
         )
         xpu_post_process(sampled_token_ids=sampled_token_ids,
                          model_output=model_output_data)