fix; refactor codes

Author: JackTan25

rtp_llm/config/gpt_init_model_parameters.py

@@ -16,6 +16,7 @@
     StaticConfig,
     get_env_bool,
     get_env_int,
+    get_env_optional_bool,
     get_env_str,
 )
 from rtp_llm.config.quant_config import (
@@ -537,6 +538,8 @@ def update_gpt_init_params_from_env(
     ):
 
         # ParallelismDistributedConfig
+        # USE_ALL_GATHER: Enable all-gather communication for pure TP (ep_size == tp_size).
+        # When enabled, DeepEP should not be used. Default is False.
         # Calculate use_all_gather: (USE_ALL_GATHER env is True) and (ep_size == tp_size)
         use_all_gather_env = get_env_bool("USE_ALL_GATHER", True)
         use_all_gather = use_all_gather_env and (
@@ -689,10 +692,24 @@ def update_gpt_init_params_from_env(
         )
 
         # MoeConfig
+        use_deepep_moe_env = get_env_optional_bool("USE_DEEPEP_MOE")
+        use_deepep_internode_env = get_env_optional_bool("USE_DEEPEP_INTERNODE")
+        use_deepep_low_latency_env = get_env_optional_bool("USE_DEEPEP_LOW_LATENCY")
+
         self.gpt_init_params.moe_config = MoeConfig(
-            use_deepep_moe=get_env_bool("USE_DEEPEP_MOE", False),
-            use_deepep_internode=get_env_bool("USE_DEEPEP_INTERNODE", False),
-            use_deepep_low_latency=get_env_bool("USE_DEEPEP_LOW_LATENCY", True),
+            use_deepep_moe=(
+                use_deepep_moe_env if use_deepep_moe_env is not None else False
+            ),
+            use_deepep_internode=(
+                use_deepep_internode_env
+                if use_deepep_internode_env is not None
+                else False
+            ),
+            use_deepep_low_latency=(
+                use_deepep_low_latency_env
+                if use_deepep_low_latency_env is not None
+                else True
+            ),
             use_deepep_p2p_low_latency=get_env_bool(
                 "USE_DEEPEP_P2P_LOW_LATENCY", False
             ),

rtp_llm/config/py_config_modules.py

@@ -40,6 +40,17 @@ def get_env_bool(name: str, default: bool = False):
     return v.lower() == "1" or v.lower() == "on" or v.lower() == "true"
 
 
+def get_env_optional_bool(name: str):
+    """
+    Get optional bool from environment variable.
+    Returns None if environment variable is not set, otherwise returns bool value.
+    """
+    v = os.environ.get(name, None)
+    if v is None or v == "":
+        return None
+    return v.lower() == "1" or v.lower() == "on" or v.lower() == "true"
+
+
 class ServerConfig:
     def __init__(self):
         self.frontend_server_count = 4
@@ -882,6 +893,25 @@ def update_from_env(self):
         self.py_hw_kernel_config.update_from_env()
         logging.info(self.to_string())
 
+    def should_auto_configure_deepep(self) -> bool:
+        """
+        Check if DeepEP should be auto-configured.
+        Returns True if environment variables are not set (None), meaning user hasn't manually configured.
+        Returns False if user has manually set any of the DeepEP environment variables.
+        """
+        use_deepep_moe_env = get_env_optional_bool("USE_DEEPEP_MOE")
+        use_deepep_internode_env = get_env_optional_bool("USE_DEEPEP_INTERNODE")
+        use_deepep_low_latency_env = get_env_optional_bool("USE_DEEPEP_LOW_LATENCY")
+
+        # Check if all environment variables are None (not set)
+        # If all are None, we should auto-configure
+        # If any is not None, user has manually configured, so we shouldn't auto-configure
+        return (
+            use_deepep_moe_env is None
+            and use_deepep_internode_env is None
+            and use_deepep_low_latency_env is None
+        )
+
     def to_string(self):
         return (
             "[server_config]\n" + self.server_config.to_string() + "\n\n"

rtp_llm/cpp/pybind/ConfigInit.cc

@@ -35,7 +35,7 @@ void register_parallelism_distributed_config(pybind11::module& m) {
              pybind11::arg("world_rank")       = 0,
              pybind11::arg("local_world_size") = 1,
              pybind11::arg("ffn_sp_size")      = 1,
-             pybind11::arg("use_all_gather")   = false)
+             pybind11::arg("use_all_gather")   = true)
         .def("to_string", &ParallelismDistributedConfig::to_string)
         .def("update_from_env", &ParallelismDistributedConfig::update_from_env_for_test)
         .def_readwrite("tp_size", &ParallelismDistributedConfig::tp_size)

rtp_llm/server/server_args/moe_group_args.py

@@ -10,24 +10,24 @@ def init_moe_group_args(parser):
         "--use_deepep_moe",
         env_name="USE_DEEPEP_MOE",
         type=str2bool,
-        default=False,
-        help="设置为 `True` 以启用 DeepEP 来处理 MoE 模型的 expert 部分。",
+        default=None,
+        help="设置为 `True` 以启用 DeepEP 来处理 MoE 模型的 expert 部分。默认值为 None，允许自动配置。",
     )
 
     moe_group.add_argument(
         "--use_deepep_internode",
         env_name="USE_DEEPEP_INTERNODE",
         type=str2bool,
-        default=False,
-        help="设置为 `True` 以启用 DeepEP 来优化跨节点 (inter-node) 通信。",
+        default=None,
+        help="设置为 `True` 以启用 DeepEP 来优化跨节点 (inter-node) 通信。默认值为 None，允许自动配置。",
     )
 
     moe_group.add_argument(
         "--use_deepep_low_latency",
         env_name="USE_DEEPEP_LOW_LATENCY",
         type=str2bool,
-        default=True,
-        help="设置为 `True` 以启用 DeepEP 的低延迟模式。",
+        default=None,
+        help="设置为 `True` 以启用 DeepEP 的低延迟模式。默认值为 None，允许自动配置。",
     )
 
     moe_group.add_argument(

rtp_llm/server/server_args/test/server_args_test.py

@@ -115,9 +115,13 @@ def test_default_args_env(self):
         self.assertIsNone(env.get("ACEXT_GEMM_CONFIG_DIR"))
 
         # 9. MOE 专家并行
-        self.assertEqual(env.get("USE_DEEPEP_MOE"), "0")
-        self.assertEqual(env.get("USE_DEEPEP_INTERNODE"), "0")
-        self.assertEqual(env.get("USE_DEEPEP_LOW_LATENCY"), "1")
+        self.assertIsNone(env.get("USE_DEEPEP_MOE"))  # 默认值为 None，允许自动配置
+        self.assertIsNone(
+            env.get("USE_DEEPEP_INTERNODE")
+        )  # 默认值为 None，允许自动配置
+        self.assertIsNone(
+            env.get("USE_DEEPEP_LOW_LATENCY")
+        )  # 默认值为 None，允许自动配置
         self.assertEqual(env.get("USE_DEEPEP_P2P_LOW_LATENCY"), "0")
         self.assertEqual(env.get("DEEP_EP_NUM_SM"), "0")
         self.assertEqual(env.get("FAKE_BALANCE_EXPERT"), "0")
@@ -945,8 +949,10 @@ def test_all_args_set_env(self):
         self.assertEqual(env["DASHSCOPE_HTTP_URL"], "http://test.url")
         self.assertEqual(env["DASHSCOPE_WEBSOCKET_URL"], "ws://test.url")
         self.assertEqual(env["OPENAI_API_KEY"], "test_openai_key")
-        self.assertEqual(env["JSON_MODEL_OVERRIDE_ARGS"],
-                         '{"rope_scaling":{"type":"yarn","factor":2.0,"original_max_position_embeddings":32768,"beta_slow":1.0,"beta_fast":1.0,"mscale":1.0,"extrapolation_factor":1.0}}')
+        self.assertEqual(
+            env["JSON_MODEL_OVERRIDE_ARGS"],
+            '{"rope_scaling":{"type":"yarn","factor":2.0,"original_max_position_embeddings":32768,"beta_slow":1.0,"beta_fast":1.0,"mscale":1.0,"extrapolation_factor":1.0}}',
+        )
 
         # 27. Lora Configuration
         self.assertEqual(env["LORA_INFO"], '{"lora1": "/path/to/lora1"}')
@@ -966,9 +972,7 @@ def test_all_args_set_env(self):
 
         # 30. Miscellaneous Configuration
         self.assertEqual(env["DISABLE_PDL"], "1")
-        self.assertEqual(
-            env["AUX_STRING"], ""
-        )
+        self.assertEqual(env["AUX_STRING"], "")
 
         # 31. PD-Separation Configuration
         self.assertEqual(env["PREFILL_RETRY_TIMES"], "2")

rtp_llm/server/server_args/util.py

@@ -2,6 +2,8 @@
 
 
 def str2bool(v):
+    if v is None:
+        return None
     if isinstance(v, bool):
         return v
     if v.lower() in ("yes", "true", "t", "1", "on"):

rtp_llm/start_server.py

@@ -5,6 +5,7 @@
 import sys
 import time
 import traceback
+from math import log
 
 import requests
 
@@ -153,41 +154,19 @@ def monitor_and_release_process(backend_process, frontend_process):
 def should_auto_configure_deepep() -> bool:
     """
     Check if DeepEP should be auto-configured.
-    Returns True if current values match defaults (user hasn't manually set them).
-    Returns False if user has manually set any of the DeepEP values to non-default.
-
-    This function reads values from StaticConfig.moe_config and compares them with
-    default values to determine if user has manually configured.
-
-    Default values:
-    - USE_DEEPEP_MOE: False
-    - USE_DEEPEP_INTERNODE: False
-    - USE_DEEPEP_LOW_LATENCY: True
+    Returns True if environment variables are not set (None), meaning user hasn't manually configured.
+    Returns False if user has manually set any of the DeepEP environment variables.
     """
-    # Default values
-    default_use_deepep_moe = False
-    default_use_deepep_internode = False
-    default_use_deepep_low_latency = True
-
-    # Read current values from StaticConfig.moe_config
-    current_use_deepep_moe = StaticConfig.moe_config.use_deepep_moe
-    current_use_deepep_internode = StaticConfig.moe_config.use_deepep_internode
-    current_use_deepep_low_latency = StaticConfig.moe_config.use_deepep_low_latency
-
-    # Check if current values match defaults
-    # If all match defaults, user hasn't manually set them, so we should auto-configure
-    # If any value differs from default, user has manually configured, so we shouldn't auto-configure
-    return (
-        current_use_deepep_moe == default_use_deepep_moe
-        and current_use_deepep_internode == default_use_deepep_internode
-        and current_use_deepep_low_latency == default_use_deepep_low_latency
-    )
+    return StaticConfig.should_auto_configure_deepep()
 
 
 def auto_configure_deepep(args: argparse.Namespace):
     """
     Automatically configure DeepEP settings based on deployment scenario.
 
+    Note: USE_ALL_GATHER should be enabled for pure TP scenarios (ep_size == tp_size).
+    When USE_ALL_GATHER is enabled, DeepEP should not be used.
+
     Configuration rules (for 8-GPU machine):
     - Non-PD separation + Inference node + Single GPU (1TP): 0, 0, 0
     - Non-PD separation + Inference node + Single-node multi-GPU (>1TP): 1, 0, 0
@@ -199,10 +178,19 @@ def auto_configure_deepep(args: argparse.Namespace):
     - PD separation + Prefill node + Multi-node multi-GPU (>=9 GPUs): 1, 0, 1
     - PD separation + Decode node + Multi-node multi-GPU (>=9 GPUs): 1, 1, 1
     """
-    # If USE_ALL_GATHER is enabled, disable all DeepEP settings
-    use_all_gather = StaticConfig.parallelism_distributed_config.use_all_gather
+    logging.info("auto configure deepep work")
+    # Get parallelism info for use_all_gather calculation
+    world_size = g_parallel_info.world_size
+    tp_size = g_parallel_info.tp_size
+    ep_size = g_parallel_info.ep_size
+    logging.info(f"world_size: {world_size}, tp_size: {tp_size}, ep_size: {ep_size}")
+    # If USE_ALL_GATHER is enabled (for pure TP scenarios), disable all DeepEP settings
+    # Calculate use_all_gather: (USE_ALL_GATHER env is True) and (ep_size == tp_size)
+    use_all_gather_env = StaticConfig.parallelism_distributed_config.use_all_gather
+    use_all_gather = use_all_gather_env and (ep_size == tp_size)
 
     if use_all_gather:
+        logging.info("use all gather in `auto_configure_deepep`")
         os.environ["USE_DEEPEP_MOE"] = "0"
         os.environ["USE_DEEPEP_LOW_LATENCY"] = "0"
         os.environ["USE_DEEPEP_INTERNODE"] = "0"
@@ -217,8 +205,6 @@ def auto_configure_deepep(args: argparse.Namespace):
     role_type = (
         role_type_enum.name if hasattr(role_type_enum, "name") else str(role_type_enum)
     )
-    world_size = g_parallel_info.world_size
-    tp_size = g_parallel_info.tp_size
 
     # Get number of nodes
     try: