Merge branch 'develop' into mm_structred_output

Author: kevincheng2

fastdeploy/engine/engine.py

@@ -946,8 +946,8 @@ def _exit_sub_services(self):
 
     def _setting_environ_variables(self):
         """
-       配置环境变量
-       """
+        配置环境变量
+        """
         variables = {
             "PADDLE_TRAINER_ID": 0,
             "PADDLE_TRAINERS_NUM": 1,

fastdeploy/import_ops.py

@@ -15,7 +15,6 @@
 import functools
 import importlib
 import inspect
-import os
 
 import paddle
 
@@ -77,7 +76,13 @@ def wrap_unified_op(original_cpp_ext_op, original_custom_op):
     @functools.wraps(original_custom_op)
     def unified_op(*args, **kwargs):
         if paddle.in_dynamic_mode():
-            return original_cpp_ext_op(*args, **kwargs)
+            res = original_cpp_ext_op(*args, **kwargs)
+            if res is None:
+                return None
+            # TODO(DrRyanHuang): Remove this if when we align the implementation of custom op and C++ extension
+            if isinstance(res, list) and len(res) == 1:
+                return res[0]
+            return res
         return original_custom_op(*args, **kwargs)
 
     return unified_op
@@ -93,17 +98,13 @@ def preprocess_static_op(global_ns):
     """
     static_op_prefix = "static_op_"
     static_op_names = [k for k in global_ns if k.startswith(static_op_prefix)]
-    enforce_eager = int(os.getenv("FD_ENFORCE_EAGER", "0")) == 1
-
-    for static_op in static_op_names:
-        op_name = static_op[len(static_op_prefix):]
-        has_dynamic_op = op_name in global_ns
-
-        if has_dynamic_op:
-            if not enforce_eager:
-                original_cpp_ext_op = global_ns[op_name]
-                original_custom_op = global_ns[static_op]
-                global_ns[op_name] = wrap_unified_op(original_cpp_ext_op,
-                                                     original_custom_op)
-        else:
-            global_ns[op_name] = global_ns[static_op]
+
+    for static_op_name in static_op_names:
+        op_name = static_op_name.removeprefix(static_op_prefix)
+        if op_name not in global_ns:
+            global_ns[op_name] = global_ns[static_op_name]
+            continue
+
+        original_cpp_ext_op = global_ns[op_name]
+        original_custom_op = global_ns[static_op_name]
+        global_ns[op_name] = wrap_unified_op(original_cpp_ext_op, original_custom_op)

fastdeploy/model_executor/layers/attention/append_attn_backend.py

@@ -91,7 +91,7 @@ def __init__(self, fd_config: FDConfig, kv_num_heads: int, num_heads: int,
         self.use_speculate: bool = self.speculative_method is not None
         self.speculate_max_draft_token_num: int = fd_config.speculative_config.num_speculative_tokens
         self.keep_pd_step_flag: bool = fd_config.speculative_config.model_type == "mtp"
-        self.rank: int = fd_config.parallel_config.tensor_parallel_rank
+        self.rank: int = fd_config.parallel_config.expert_parallel_rank * fd_config.parallel_config.tensor_parallel_degree + fd_config.parallel_config.tensor_parallel_rank
 
         self.kv_num_heads: int = kv_num_heads
         self.num_heads: int = num_heads
@@ -108,12 +108,12 @@ def __init__(self, fd_config: FDConfig, kv_num_heads: int, num_heads: int,
 
         if fd_config.parallel_config.expert_parallel_rank is None:
             fd_config.parallel_config.expert_parallel_rank = 0
-        device_id = self.rank + fd_config.parallel_config.tensor_parallel_degree * \
-            fd_config.parallel_config.expert_parallel_rank
         if self.device_id is None:
-            self.device_id = device_id
+            self.device_id = self.rank
         else:
-            self.device_id = self.device_id.split(",")[device_id]
+            device_ids = self.device_id.split(",")
+            rank_index = self.rank % len(device_ids)
+            self.device_id = self.device_id[rank_index]
 
     def init_attention_metadata(self, forward_meta: ForwardMeta):
         """Initialize attntion metadata hence all layers in the forward pass can reuse it."""

fastdeploy/model_executor/layers/attention/flash_attn_backend.py

@@ -100,7 +100,7 @@ def __init__(self, fd_config: FDConfig, kv_num_heads: int, num_heads: int,
         self.use_speculate = self.speculative_method is not None
         self.speculate_max_draft_token_num = fd_config.speculative_config.num_speculative_tokens
         self.keep_pd_step_flag: bool = fd_config.speculative_config.model_type == "mtp"
-        self.rank: int = fd_config.parallel_config.tensor_parallel_rank
+        self.rank: int = fd_config.parallel_config.expert_parallel_rank * fd_config.parallel_config.tensor_parallel_degree + fd_config.parallel_config.tensor_parallel_rank
 
         # pd_disaggregation
         self.use_pd_disaggregation: int = int(
@@ -110,12 +110,13 @@ def __init__(self, fd_config: FDConfig, kv_num_heads: int, num_heads: int,
 
         if fd_config.parallel_config.expert_parallel_rank is None:
             fd_config.parallel_config.expert_parallel_rank = 0
-        device_id = self.rank + fd_config.parallel_config.tensor_parallel_degree * \
-            fd_config.parallel_config.expert_parallel_rank
+            
         if self.device_id is None:
-            self.device_id = device_id
+            self.device_id = self.rank
         else:
-            self.device_id = self.device_id.split(",")[device_id]
+            device_ids = self.device_id.split(",")
+            rank_index = self.rank % len(device_ids)
+            self.device_id = self.device_id[rank_index]
 
     def get_attntion_meta(self):
         """get_attntion_meta"""

fastdeploy/model_executor/layers/attention/mla_attention_backend.py

@@ -109,7 +109,7 @@ def __init__(self, fd_config: FDConfig, kv_num_heads: int, num_heads: int,
         self.use_speculate: bool = self.speculative_method is not None
         self.speculate_max_draft_token_num: int = fd_config.speculative_config.num_speculative_tokens
         self.keep_pd_step_flag: bool = fd_config.speculative_config.model_type == "mtp"
-        self.rank: int = fd_config.parallel_config.tensor_parallel_rank
+        self.rank: int = fd_config.parallel_config.expert_parallel_rank * fd_config.parallel_config.tensor_parallel_degree + fd_config.parallel_config.tensor_parallel_rank
 
         self.kv_num_heads: int = kv_num_heads
         self.num_heads: int = num_heads
@@ -135,10 +135,13 @@ def __init__(self, fd_config: FDConfig, kv_num_heads: int, num_heads: int,
             os.getenv("FLAGS_use_pd_disaggregation", 0))
         self.start_layer_index: int = fd_config.model_config.start_layer_index
         self.device_id: int = os.getenv("CUDA_VISIBLE_DEVICES", None)
+        
         if self.device_id is None:
             self.device_id = self.rank
         else:
-            self.device_id = self.device_id.split(",")[self.rank]
+            device_ids = self.device_id.split(",")
+            rank_index = self.rank % len(device_ids)
+            self.device_id = self.device_id[rank_index]
 
     def init_attention_metadata(self, forward_meta: ForwardMeta):
         """Initialize attention metadata hence all layers in the forward pass can reuse it."""

fastdeploy/model_executor/layers/attention/xpu_attn_backend.py

@@ -91,7 +91,7 @@ def __init__(self, fd_config: FDConfig, kv_num_heads: int, num_heads: int,
         # self.use_speculate = self.speculate_method is not None
         # self.speculate_max_draft_token_num = fd_config.parallel_config.speculate_max_draft_tokens
         self.keep_pd_step_flag: bool = fd_config.speculative_config.model_type == "mtp"
-        self.rank: int = fd_config.parallel_config.tensor_parallel_rank
+        self.rank: int = fd_config.parallel_config.expert_parallel_rank * fd_config.parallel_config.tensor_parallel_degree + fd_config.parallel_config.tensor_parallel_rank
 
         self.kv_num_heads: int = kv_num_heads
         self.num_heads: int = num_heads

fastdeploy/model_executor/models/ernie4_5_vl/ernie4_5_vl_moe.py

@@ -445,7 +445,7 @@ def forward(
             forward_meta.seq_lens_this_time,
             forward_meta.cu_seqlens_q,
             score_text,
-        )[0].cast(self._dtype)
+        ).cast(self._dtype)
         # -----------------------
 
         out = self.norm(hidden_states)

test/ci_use/EB_VL_Lite/test_EB_VL_Lite_serving.py

@@ -176,58 +176,31 @@ def consistent_payload():
         "seed": 13  # fixed random seed
     }
 
-# ==========================
-# Helper function to calculate difference rate between two texts
-# ==========================
-def calculate_diff_rate(text1, text2):
-    """
-    Calculate the difference rate between two strings
-    based on the normalized Levenshtein edit distance.
-    Returns a float in [0,1], where 0 means identical.
-    """
-    if text1 == text2:
-        return 0.0
-
-    len1, len2 = len(text1), len(text2)
-    dp = [[0] * (len2 + 1) for _ in range(len1 + 1)]
-
-    for i in range(len1 + 1):
-        for j in range(len2 + 1):
-            if i == 0 or j == 0:
-                dp[i][j] = i + j
-            elif text1[i - 1] == text2[j - 1]:
-                dp[i][j] = dp[i - 1][j - 1]
-            else:
-                dp[i][j] = 1 + min(dp[i - 1][j], dp[i][j - 1], dp[i - 1][j - 1])
-
-    edit_distance = dp[len1][len2]
-    max_len = max(len1, len2)
-    return edit_distance / max_len if max_len > 0 else 0.0
 
 # ==========================
 # Consistency test for repeated runs with fixed payload
 # ==========================
 def test_consistency_between_runs(api_url, headers, consistent_payload):
     """
-    Test that two runs with the same fixed input produce similar outputs.
+    Test that result is same as the base result.
     """
-    # First request
+    # request
     resp1 = requests.post(api_url, headers=headers, json=consistent_payload)
     assert resp1.status_code == 200
     result1 = resp1.json()
     content1 = result1["choices"][0]["message"]["content"]
 
-    # Second request
-    resp2 = requests.post(api_url, headers=headers, json=consistent_payload)
-    assert resp2.status_code == 200
-    result2 = resp2.json()
-    content2 = result2["choices"][0]["message"]["content"]
-
-    # Calculate difference rate
-    diff_rate = calculate_diff_rate(content1, content2)
+    # base result
+    base_path = os.getenv("MODEL_PATH")
+    if base_path:
+        base_file = os.path.join(base_path, "ernie-4_5-vl-base")
+    else:
+        base_file = "ernie-4_5-vl-base"
+    with open(base_file, "r") as f:
+        content2 = f.read()
 
-    # Verify that the difference rate is below the threshold
-    assert diff_rate < 0.05, "Output difference too large ({:.4%})".format(diff_rate)
+    # Verify that result is same as the base result
+    assert content1 == content2
 
 # ==========================
 # OpenAI Client Chat Completion Test
@@ -322,4 +295,4 @@ def test_streaming_chat(openai_client, capsys):
     for chunk in response:
         if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
             output.append(chunk.choices[0].delta.content)
-    assert len(output) > 2
+    assert len(output) > 2