diff --git a/lmdeploy/cli/serve.py b/lmdeploy/cli/serve.py
index 47c16bfe52..6bffe54777 100644
--- a/lmdeploy/cli/serve.py
+++ b/lmdeploy/cli/serve.py
@@ -130,7 +130,7 @@ def add_parser_api_server():
         hf_overrides = ArgumentHelper.hf_overrides(pt_group)
         disable_metrics = ArgumentHelper.disable_metrics(pt_group)
         dp = ArgumentHelper.dp(pt_group)
-        ArgumentHelper.ep(pt_group)
+        ep_act = ArgumentHelper.ep(pt_group)
         ArgumentHelper.enable_microbatch(pt_group)
         ArgumentHelper.enable_eplb(pt_group)
         ArgumentHelper.role(pt_group)
@@ -158,6 +158,7 @@ def add_parser_api_server():
         tb_group._group_actions.append(hf_overrides)
         tb_group._group_actions.append(disable_metrics)
         tb_group._group_actions.append(dp)
+        tb_group._group_actions.append(ep_act)
         tb_group._group_actions.append(language_model_only)
         ArgumentHelper.cp(tb_group)
         ArgumentHelper.rope_scaling_factor(tb_group)
@@ -270,6 +271,7 @@ def api_server(args):
                                                    tp=args.tp,
                                                    dp=args.dp,
                                                    cp=args.cp,
+                                                   ep=args.ep,
                                                    nnodes=args.nnodes,
                                                    node_rank=args.node_rank,
                                                    dist_init_addr=args.dist_init_addr,
diff --git a/lmdeploy/messages.py b/lmdeploy/messages.py
index b6dce131ee..e07a46ae5e 100644
--- a/lmdeploy/messages.py
+++ b/lmdeploy/messages.py
@@ -304,6 +304,7 @@ class TurbomindEngineConfig:
     tp: int = 1
     dp: int = 1
     cp: int = 1
+    ep: int = 1
     device_num: int = None
     attn_tp_size: int = None
     attn_cp_size: int = None
@@ -345,6 +346,7 @@ def __post_init__(self):
         """Check input validation."""
         assert self.dtype in ['auto', 'float16', 'bfloat16']
         assert self.tp >= 1, 'tp must be a positive integer'
+        assert self.ep >= 1, 'ep must be a positive integer'
         assert self.cache_max_entry_count > 0, 'invalid cache_max_entry_count'
         try:
             self.quant_policy = QuantPolicy(self.quant_policy)
diff --git a/lmdeploy/turbomind/builders/_base.py b/lmdeploy/turbomind/builders/_base.py
index b332b2b7c7..023c02faa8 100644
--- a/lmdeploy/turbomind/builders/_base.py
+++ b/lmdeploy/turbomind/builders/_base.py
@@ -142,10 +142,33 @@ class Context:
     def __init__(self, devices, data_type):
         self.devices = devices
         self.data_type = data_type
+        self._active_mask_stack = [(True,) * len(devices)]
+
+    @property
+    def active_mask(self):
+        return self._active_mask_stack[-1]
+
+    def active_mask_scope(self, active_mask):
+        return _ActiveMaskScope(self, active_mask)
+
+
+class _ActiveMaskScope:
+    """Temporarily restrict builders created under the context."""
+
+    def __init__(self, ctx, active_mask):
+        self._ctx = ctx
+        self._active_mask = tuple(bool(x) for x in active_mask)
+
+    def __enter__(self):
+        self._ctx._active_mask_stack.append(self._active_mask)
+
+    def __exit__(self, exc_type, exc, tb):
+        self._ctx._active_mask_stack.pop()
+        return False
 
 
 class ParallelGroup:
-    """Bundle a parallelism size with per-GPU TP ranks."""
+    """Bundle a parallelism size with per-GPU ranks."""
     def __init__(self, size, ranks):
         self.size = size
         self.ranks = ranks
@@ -181,6 +204,7 @@ def __init__(self, config, ctx):
         # __setattr__.
         self._built = False
         self._ctx = ctx
+        self._active_mask = ctx.active_mask
         self.tp = ParallelGroup(1, None)   # default: no TP
         self.config = config
         if hasattr(self.config, 'data_type'):
@@ -220,6 +244,9 @@ def _rank_for(self, gpu_idx: int) -> int:
             return self.tp.ranks[gpu_idx]
         return 0
 
+    def _is_active(self, gpu_idx: int) -> bool:
+        return self._active_mask[gpu_idx]
+
     # ------------------------------------------------------------------
     # Add methods — stage into pending dicts (pre-build only)
     # ------------------------------------------------------------------
@@ -280,6 +307,9 @@ def _add_linear(self, name: str, linear: Linear,
         # --- Per-GPU: standalone creation + tensor copy --------------------
         handles = []
         for i, ctx in enumerate(self._ctx.devices):
+            if not self._is_active(i):
+                handles.append(None)
+                continue
             with ctx:
                 rank = self._rank_for(i) if tp > 1 else 0
 
@@ -363,6 +393,9 @@ def _create_handles(self):
         """Create one C++ module per context via ``_tm.create_module(cfg)``."""
         handles = []
         for i, ctx in enumerate(self._ctx.devices):
+            if not self._is_active(i):
+                handles.append(None)
+                continue
             with ctx:
                 cfg = self._cfg_for_rank(i)
                 handle = _tm.create_module(cfg)
@@ -380,7 +413,9 @@ def _cfg_for_rank(self, gpu_idx: int):
     def _commit_child(self, name: str, handles: list):
         """Attach pre-created per-GPU child handles to parent handles."""
         for i, (parent_h, child_h) in enumerate(
-                zip(self._handles, handles)):
+                zip(self._handles, handles, strict=True)):
+            if parent_h is None or child_h is None:
+                continue
             with self._ctx.devices[i]:
                 parent_h.add_child_raw(name, child_h)
 
@@ -405,6 +440,8 @@ def _commit_tensor(self, name: str, tensor: torch.Tensor,
         split_dim = _SPLIT_SIDE_TO_DIM.get(split_side) if split_side else None
 
         for i, handle in enumerate(self._handles):
+            if handle is None:
+                continue
             with self._ctx.devices[i]:
                 rank = self._rank_for(i) if tp > 1 else 0
                 shard = _shard(tensor, split_dim, tp, rank)
diff --git a/lmdeploy/turbomind/builders/moe.py b/lmdeploy/turbomind/builders/moe.py
index a4a1bcb36d..fe350d6006 100644
--- a/lmdeploy/turbomind/builders/moe.py
+++ b/lmdeploy/turbomind/builders/moe.py
@@ -1,7 +1,7 @@
 # Copyright (c) OpenMMLab. All rights reserved.
 from __future__ import annotations
 
-from ._base import Builder, SplitSide
+from ._base import Builder, ParallelGroup, SplitSide
 
 # ---------------------------------------------------------------------------
 # MoeBuilder -- gate, non-expert params
@@ -11,6 +11,24 @@
 class MoeBuilder(Builder):
     """MoE weight loading builder."""
 
+    def __init__(self, config, ctx, ep: ParallelGroup | None = None):
+        super().__init__(config, ctx)
+        self.ep = ep or ParallelGroup(1, None)
+        if self.ep.size > 1 and config.expert_num % self.ep.size != 0:
+            raise ValueError(
+                f'num_experts={config.expert_num} must be divisible by '
+                f'ep={self.ep.size}')
+        self.config.ep_size = self.ep.size
+
+    def _cfg_for_rank(self, gpu_idx: int):
+        """Set this GPU context's EP rank when EP is active."""
+        if self.ep.size <= 1:
+            return super()._cfg_for_rank(gpu_idx)
+        else:
+            cfg = self.config.clone()
+            cfg.ep_rank = self.ep.ranks[gpu_idx]
+            return cfg
+
     def add_gate(self, name, linear):
         """Commit a gate linear (broadcast, no split)."""
         self._add_linear(name, linear, split_side=None)
@@ -20,3 +38,21 @@ def add_param(self, name, tensor, split_side=None):
         if split_side is not None and not isinstance(split_side, SplitSide):
             split_side = None  # specs may pass None for broadcast
         self._add_tensor(name, tensor, split_side)
+
+    def range(self, num_experts):
+        for e in range(num_experts):
+            active_mask = self._expert_active_mask(e)
+            if not any(active_mask):
+                continue
+            with self._ctx.active_mask_scope(active_mask):
+                yield e
+
+    def _expert_active_mask(self, expert_idx: int):
+        ep_size = self.ep.size
+        if ep_size <= 1:
+            return [True] * len(self._ctx.devices)
+        ranks = self.ep.ranks
+        assert ranks is not None
+        local = self.config.expert_num // ep_size
+        return [rank * local <= expert_idx < (rank + 1) * local
+                for rank in ranks]
diff --git a/lmdeploy/turbomind/model_loader.py b/lmdeploy/turbomind/model_loader.py
index c7f8851f01..d79406b456 100644
--- a/lmdeploy/turbomind/model_loader.py
+++ b/lmdeploy/turbomind/model_loader.py
@@ -36,6 +36,8 @@ def _bind_runtime(self):
                                 [mc.attn_tp_rank(g) for g in range(self.gpu_count)])
         mlp_tp = ParallelGroup(ec.mlp_tp_size,
                                [mc.mlp_tp_rank(g) for g in range(self.gpu_count)])
+        ep = ParallelGroup(ec.ep,
+                           [mc.ep_rank(g) for g in range(self.gpu_count)])
         model_tp = ParallelGroup(ec.attn_tp_size * ec.attn_cp_size,
                                  [mc.model_tp_rank(g) for g in range(self.gpu_count)])
 
@@ -44,6 +46,7 @@ def _bind_runtime(self):
             root_handles=[mc.root(g) for g in range(self.gpu_count)],
             attn_tp=attn_tp,
             mlp_tp=mlp_tp,
+            ep=ep,
             model_tp=model_tp,
         )
 
diff --git a/lmdeploy/turbomind/models/glm4_moe_lite.py b/lmdeploy/turbomind/models/glm4_moe_lite.py
index e779633460..76075ff051 100644
--- a/lmdeploy/turbomind/models/glm4_moe_lite.py
+++ b/lmdeploy/turbomind/models/glm4_moe_lite.py
@@ -115,7 +115,7 @@ def ffn(self, pfx, inter_size, is_expert=False):
     def moe(self, pfx):
         cfg = self._moe_cfg.clone()
 
-        m = MoeBuilder(cfg, self._ctx)
+        m = MoeBuilder(cfg, self._ctx, ep=self._ep)
 
         m.add_gate('gate', self._linear(pfx + 'gate'))
 
@@ -123,7 +123,7 @@ def moe(self, pfx):
         m.add_param('score_correction_bias', correction)
 
         experts = ModuleListBuilder(ModuleListConfig(), self._ctx)
-        for e in range(cfg.expert_num):
+        for e in m.range(cfg.expert_num):
             experts[e] = self.ffn(pfx + 'experts' + e,
                                   self.cfg.moe_intermediate_size, is_expert=True)
         m.experts = experts.build()
diff --git a/lmdeploy/turbomind/models/gpt_oss.py b/lmdeploy/turbomind/models/gpt_oss.py
index 5480eaee7b..018c4d4ba8 100644
--- a/lmdeploy/turbomind/models/gpt_oss.py
+++ b/lmdeploy/turbomind/models/gpt_oss.py
@@ -107,11 +107,11 @@ def reorder(x):
 
     def moe(self, pfx):
         cfg = self._moe_cfg.clone()
-        m = MoeBuilder(cfg, self._ctx)
+        m = MoeBuilder(cfg, self._ctx, ep=self._ep)
         m.add_gate('gate', self._linear(pfx + 'router'))
         experts_pfx = pfx + 'experts'
         experts = ModuleListBuilder(ModuleListConfig(), self._ctx)
-        for e in range(cfg.expert_num):
+        for e in m.range(cfg.expert_num):
             experts[e] = self._packed_moe_ffn(experts_pfx, e)
         m.experts = experts.build()
         return m.build()
diff --git a/lmdeploy/turbomind/models/internvl.py b/lmdeploy/turbomind/models/internvl.py
index ee3f344850..022f9d91f8 100644
--- a/lmdeploy/turbomind/models/internvl.py
+++ b/lmdeploy/turbomind/models/internvl.py
@@ -60,12 +60,13 @@ def __init__(self, cfg: PretrainedConfig, *, resolver):
         self.vision_model = None
 
     def bind_runtime(self, *, ctx, root_handles,
-                     attn_tp, mlp_tp, model_tp):
+                     attn_tp, mlp_tp, ep, model_tp):
         self.text_model.bind_runtime(
             ctx=ctx,
             root_handles=root_handles,
             attn_tp=attn_tp,
             mlp_tp=mlp_tp,
+            ep=ep,
             model_tp=model_tp,
         )
 
diff --git a/lmdeploy/turbomind/models/mixtral.py b/lmdeploy/turbomind/models/mixtral.py
index f6c41b4e3a..07c90aa6a9 100644
--- a/lmdeploy/turbomind/models/mixtral.py
+++ b/lmdeploy/turbomind/models/mixtral.py
@@ -83,11 +83,11 @@ def ffn(self, pfx, is_expert=False):
     def moe(self, pfx):
         cfg = self._moe_cfg.clone()
 
-        m = MoeBuilder(cfg, self._ctx)
+        m = MoeBuilder(cfg, self._ctx, ep=self._ep)
         m.add_gate('gate', self._linear(pfx + 'gate'))
 
         experts = ModuleListBuilder(ModuleListConfig(), self._ctx)
-        for e in range(self._n_experts):
+        for e in m.range(self._n_experts):
             experts[e] = self.ffn(pfx + 'experts' + e, is_expert=True)
         m.experts = experts.build()
 
diff --git a/lmdeploy/turbomind/models/qwen2.py b/lmdeploy/turbomind/models/qwen2.py
index 64b5accb92..c23c767afa 100644
--- a/lmdeploy/turbomind/models/qwen2.py
+++ b/lmdeploy/turbomind/models/qwen2.py
@@ -110,12 +110,12 @@ def ffn(self, pfx, inter_size, is_expert=False):
     def moe(self, pfx):
         cfg = self._moe_cfg.clone()
 
-        m = MoeBuilder(cfg, self._ctx)
+        m = MoeBuilder(cfg, self._ctx, ep=self._ep)
 
         m.add_gate('gate', self._linear(pfx + 'gate'))
 
         experts = ModuleListBuilder(ModuleListConfig(), self._ctx)
-        for e in range(self.cfg.num_experts):
+        for e in m.range(self.cfg.num_experts):
             experts[e] = self.ffn(pfx + 'experts' + e,
                                   self.cfg.moe_intermediate_size,
                                   is_expert=True)
diff --git a/lmdeploy/turbomind/models/qwen3.py b/lmdeploy/turbomind/models/qwen3.py
index f30f1d95b8..98fcb8eee1 100644
--- a/lmdeploy/turbomind/models/qwen3.py
+++ b/lmdeploy/turbomind/models/qwen3.py
@@ -110,11 +110,11 @@ def ffn(self, pfx, is_expert=False):
     def moe(self, pfx):
         cfg = self._moe_cfg.clone()
 
-        m = MoeBuilder(cfg, self._ctx)
+        m = MoeBuilder(cfg, self._ctx, ep=self._ep)
         m.add_gate('gate', self._linear(pfx + 'gate'))
 
         experts = ModuleListBuilder(ModuleListConfig(), self._ctx)
-        for e in range(self.cfg.num_experts):
+        for e in m.range(self.cfg.num_experts):
             experts[e] = self.ffn(pfx + 'experts' + e, is_expert=True)
         m.experts = experts.build()
 
diff --git a/lmdeploy/turbomind/models/qwen3_5.py b/lmdeploy/turbomind/models/qwen3_5.py
index bde8ff8e9d..f820d7455a 100644
--- a/lmdeploy/turbomind/models/qwen3_5.py
+++ b/lmdeploy/turbomind/models/qwen3_5.py
@@ -206,13 +206,13 @@ def ffn(self, pfx, inter_size, is_expert=False):
     def moe(self, pfx):
         cfg = self._moe_cfg.clone()
 
-        m = MoeBuilder(cfg, self._ctx)
+        m = MoeBuilder(cfg, self._ctx, ep=self._ep)
 
         m.add_gate('gate', self._linear(pfx + 'gate'))
 
         experts_pfx = pfx + 'experts'
         experts = ModuleListBuilder(ModuleListConfig(), self._ctx)
-        for e in range(self._n_experts):
+        for e in m.range(self._n_experts):
             experts[e] = self._moe_expert_ffn(
                 experts_pfx, e, self.cfg.moe_intermediate_size)
         m.experts = experts.build()
@@ -639,7 +639,7 @@ def __init__(self, cfg: Qwen3_5Config | Qwen3_5MoeConfig, *, resolver,
                 vision_cfg, resolver=vision_resolver or resolver)
 
     def bind_runtime(self, *, ctx, root_handles,
-                     attn_tp, mlp_tp, model_tp):
+                     attn_tp, mlp_tp, ep, model_tp):
         for m in (self.text_model, self.vision_model):
             if m is not None:
                 m.bind_runtime(
@@ -647,6 +647,7 @@ def bind_runtime(self, *, ctx, root_handles,
                     root_handles=root_handles,
                     attn_tp=attn_tp,
                     mlp_tp=mlp_tp,
+                    ep=ep,
                     model_tp=model_tp,
                 )
 
diff --git a/lmdeploy/turbomind/text_model.py b/lmdeploy/turbomind/text_model.py
index 9d86fd5cff..75842f6e07 100644
--- a/lmdeploy/turbomind/text_model.py
+++ b/lmdeploy/turbomind/text_model.py
@@ -40,11 +40,12 @@ def _vocab_size(self) -> int:
         return self.cfg.vocab_size
 
     def bind_runtime(self, *, ctx, root_handles,
-                     attn_tp, mlp_tp, model_tp):
+                     attn_tp, mlp_tp, ep, model_tp):
         self._ctx = ctx
         self._root_handles = root_handles
         self._attn_tp = attn_tp
         self._mlp_tp = mlp_tp
+        self._ep = ep
         self._model_tp = model_tp
 
     def _linear(self, pfx: Prefix, *,
diff --git a/lmdeploy/turbomind/turbomind.py b/lmdeploy/turbomind/turbomind.py
index 30520a86f5..a3468e9716 100644
--- a/lmdeploy/turbomind/turbomind.py
+++ b/lmdeploy/turbomind/turbomind.py
@@ -82,25 +82,28 @@ def complete_parallel_config(cfg: TurbomindEngineConfig):
 
 def update_parallel_config(cfg: TurbomindEngineConfig):
     cfg.device_num = len(cfg.devices) * cfg.nnodes if cfg.devices else cfg.device_num
+    assert cfg.ep == 1 or cfg.tp == 1
     if not complete_parallel_config(cfg):
-        total = cfg.dp * cfg.tp
+        total = cfg.dp * cfg.ep * cfg.tp
         if not cfg.device_num:
             count = torch.cuda.device_count() * cfg.nnodes
             if total < count:
                 count = total
             cfg.device_num = count
         assert total % cfg.device_num == 0
+        size = max(cfg.ep, cfg.tp)
         overlap = total // cfg.device_num
-        attn_dp_size = overlap
-        mlp_tp_size = overlap
-        inner_tp_size = cfg.tp // mlp_tp_size
-        cfg.outer_dp_size = cfg.dp // attn_dp_size
-        cfg.attn_dp_size = attn_dp_size
+        inner_tp_size = size // overlap
+        cfg.outer_dp_size = cfg.dp // overlap
+        cfg.attn_dp_size = overlap
         cfg.attn_tp_size = inner_tp_size // cfg.cp
         cfg.attn_cp_size = cfg.cp
+        comm_size = cfg.attn_dp_size * cfg.attn_tp_size * cfg.attn_cp_size
         cfg.mlp_dp_size = 1
-        cfg.mlp_tp_size = mlp_tp_size * inner_tp_size
-    assert cfg.attn_dp_size * cfg.attn_tp_size * cfg.attn_cp_size == cfg.mlp_dp_size * cfg.mlp_tp_size
+        cfg.mlp_tp_size = comm_size // cfg.ep if cfg.ep > 1 else overlap * inner_tp_size
+    if cfg.ep > 1:
+        assert cfg.nnodes == 1, 'ep > 1 is only supported in single-node mode'
+        assert cfg.mlp_tp_size == 1, 'Only support mlp_tp_size == 1 when ep > 1'
     assert cfg.attn_dp_size * cfg.attn_tp_size * cfg.attn_cp_size * cfg.outer_dp_size == cfg.device_num
     # update devices
     cfg.devices = cfg.devices or list(range(cfg.device_num // cfg.nnodes))
@@ -252,6 +255,7 @@ def _from_hf(self, model_path: str, engine_config: TurbomindEngineConfig,
         ec.attn_tp_size = engine_config.attn_tp_size
         ec.attn_cp_size = engine_config.attn_cp_size
         ec.mlp_tp_size = engine_config.mlp_tp_size
+        ec.ep_size = engine_config.ep
         ec.devices = engine_config.devices
         ec.nnodes = engine_config.nnodes
         ec.node_rank = engine_config.node_rank
diff --git a/src/turbomind/comm/device_comm.h b/src/turbomind/comm/device_comm.h
index a6948762df..9ce0be0d12 100644
--- a/src/turbomind/comm/device_comm.h
+++ b/src/turbomind/comm/device_comm.h
@@ -5,6 +5,7 @@
 #include <memory>
 
 #include <stdexcept>
+#include <vector>
 
 #include <cuda_runtime.h>
 
@@ -54,6 +55,16 @@ class DeviceCommImpl {
                            int          group,
                            cudaStream_t stream) = 0;
 
+    virtual void AllGatherV(const void*                sendbuff,
+                            void*                      recvbuff,
+                            const std::vector<size_t>& counts,
+                            DataType                   type,
+                            int                        group,
+                            cudaStream_t               stream)
+    {
+        throw std::runtime_error("not implemented");
+    }
+
     virtual void ReduceScatter(const void*  sendbuff,  //
                                void*        recvbuff,
                                size_t       recvcount,
@@ -64,6 +75,16 @@ class DeviceCommImpl {
         throw std::runtime_error("not implemented");
     }
 
+    virtual void ReduceScatterV(const void*                sendbuff,
+                                void*                      recvbuff,
+                                const std::vector<size_t>& counts,
+                                DataType                   type,
+                                int                        group,
+                                cudaStream_t               stream)
+    {
+        throw std::runtime_error("not implemented");
+    }
+
     virtual void AllreduceResidualBiasRMSnorm(void*        hidden,
                                               void*        residual,
                                               const void*  bias,
diff --git a/src/turbomind/comm/nccl/nccl.cu b/src/turbomind/comm/nccl/nccl.cu
index 67e0fef181..a7f843b657 100644
--- a/src/turbomind/comm/nccl/nccl.cu
+++ b/src/turbomind/comm/nccl/nccl.cu
@@ -44,6 +44,8 @@ static inline ncclDataType_t to_nccl_dtype(DataType type)
             return ncclBfloat16;
         case kUint8:
             return ncclUint8;
+        case kInt64:
+            return ncclInt64;
         default:
             throw std::runtime_error("not supported");
     }
@@ -262,6 +264,46 @@ public:
         NCCLCHECK(ncclGroupEnd());
     }
 
+    void AllGatherV(const void*                sendbuff,
+                    void*                      recvbuff,
+                    const std::vector<size_t>& counts,
+                    DataType                   type,
+                    int                        group,
+                    cudaStream_t               stream) override
+    {
+        const size_t         elem_size = byte_size(type);
+        const ncclDataType_t nccl_type = to_nccl_dtype(type);
+        const int            n_ranks   = this->n_ranks(group);
+        const int            rank      = this->rank(group);
+        ncclComm_t           comm      = groups_.at(group);
+
+        TM_CHECK_EQ((int)counts.size(), n_ranks);
+
+        bool equal_counts = true;
+        for (auto i = 1; i < counts.size(); ++i) {
+            if (counts[i] != counts[0]) {
+                equal_counts = false;
+                break;
+            }
+        }
+        if (equal_counts) {
+            return AllGather(sendbuff, recvbuff, counts[0], type, group, stream);
+        }
+
+        NCCLCHECK(ncclGroupStart());
+        size_t offset = 0;
+        for (int i = 0; i < n_ranks; ++i) {
+            const size_t count = counts[i];
+            if (count) {
+                auto*       recv = static_cast<char*>(recvbuff) + elem_size * offset;
+                const void* send = i == rank ? sendbuff : recv;
+                NCCLCHECK(ncclBroadcast(send, recv, count, nccl_type, i, comm, stream));
+            }
+            offset += count;
+        }
+        NCCLCHECK(ncclGroupEnd());
+    }
+
     void ReduceScatter(
         const void* sendbuff, void* recvbuff, size_t recvcount, DataType type, int group, cudaStream_t stream) override
     {
@@ -271,6 +313,46 @@ public:
         NCCLCHECK(ncclGroupEnd());
     }
 
+    void ReduceScatterV(const void*                sendbuff,
+                        void*                      recvbuff,
+                        const std::vector<size_t>& counts,
+                        DataType                   type,
+                        int                        group,
+                        cudaStream_t               stream) override
+    {
+        const size_t         elem_size = byte_size(type);
+        const ncclDataType_t nccl_type = to_nccl_dtype(type);
+        const int            n_ranks   = this->n_ranks(group);
+        const int            rank      = this->rank(group);
+        ncclComm_t           comm      = groups_.at(group);
+
+        TM_CHECK_EQ((int)counts.size(), n_ranks);
+
+        bool equal_counts = true;
+        for (auto i = 1; i < counts.size(); ++i) {
+            if (counts[i] != counts[0]) {
+                equal_counts = false;
+                break;
+            }
+        }
+        if (equal_counts) {
+            return ReduceScatter(sendbuff, recvbuff, counts[0], type, group, stream);
+        }
+
+        NCCLCHECK(ncclGroupStart());
+        size_t offset = 0;
+        for (int i = 0; i < n_ranks; ++i) {
+            const size_t count = counts[i];
+            if (count) {
+                const auto* send = static_cast<const char*>(sendbuff) + elem_size * offset;
+                auto*       recv = i == rank ? recvbuff : const_cast<char*>(send);
+                NCCLCHECK(ncclReduce(send, recv, count, nccl_type, ncclSum, i, comm, stream));
+            }
+            offset += count;
+        }
+        NCCLCHECK(ncclGroupEnd());
+    }
+
     void AllreduceResidualBiasRMSnorm(void*        hidden,
                                       void*        residual,
                                       const void*  bias,
@@ -392,7 +474,7 @@ public:
                    int          group,
                    cudaStream_t stream) override
     {
-        NCCLCHECK(ncclBroadcast(recvbuff, recvbuff, count, to_nccl_dtype(type), root, groups_.at(group), stream));
+        NCCLCHECK(ncclBroadcast(sendbuff, recvbuff, count, to_nccl_dtype(type), root, groups_.at(group), stream));
     }
 
 private:
diff --git a/src/turbomind/engine/engine_config.h b/src/turbomind/engine/engine_config.h
index 02ec4a2ff2..f7d13c029f 100644
--- a/src/turbomind/engine/engine_config.h
+++ b/src/turbomind/engine/engine_config.h
@@ -36,6 +36,7 @@ struct EngineConfig {
     X(int, attn_tp_size)                                                                                               \
     X(int, attn_cp_size)                                                                                               \
     X(int, mlp_tp_size)                                                                                                \
+    X(int, ep_size, 1)                                                                                                 \
     X(std::vector<int>, devices)                                                                                       \
     X(int, nnodes)                                                                                                     \
     X(int, node_rank)                                                                                                  \
diff --git a/src/turbomind/kernels/activation.cu b/src/turbomind/kernels/activation.cu
index cab79c6ee7..1406c9c78b 100644
--- a/src/turbomind/kernels/activation.cu
+++ b/src/turbomind/kernels/activation.cu
@@ -1,4 +1,6 @@
 
+#include <type_traits>
+
 #include "src/turbomind/core/data_type.h"
 
 #include "src/turbomind/kernels/activation.h"
@@ -26,9 +28,14 @@ struct Silu {
     }
 };
 
-template<int vec_size, class Activation, class T>
-__global__ void ActivationKernel(
-    T* gate_buf, const T* __restrict__ up_buf, Activation activation, int64_t stride, int token_num, int dim)
+template<bool has_valid_tokens, int vec_size, class Activation, class T>
+__global__ void ActivationKernel(T* gate_buf,
+                                 const T* __restrict__ up_buf,
+                                 Activation activation,
+                                 int64_t    stride,
+                                 int        token_num,
+                                 int        dim,
+                                 const int* num_valid_tokens)
 {
     if constexpr (TURBOMIND_ARCH_DTYPE_GUARD(data_type_v<T>)) {
         const int di = threadIdx.x + blockIdx.y * blockDim.x;
@@ -36,6 +43,12 @@ __global__ void ActivationKernel(
 
         dim /= vec_size;
 
+        if constexpr (has_valid_tokens) {
+            if (ti >= __ldg(num_valid_tokens)) {
+                return;
+            }
+        }
+
         if (di >= dim) {
             return;
         }
@@ -61,6 +74,12 @@ __global__ void ActivationKernel(
 }
 
 void Activation(Ref<Tensor> gate_, const Tensor& up, ActivationType type, cudaStream_t stream)
+{
+    Activation(gate_, up, type, nullptr, stream);
+}
+
+void Activation(
+    Ref<Tensor> gate_, const Tensor& up, ActivationType type, const int* num_valid_tokens, cudaStream_t stream)
 {
     auto& gate = gate_.get();
 
@@ -69,52 +88,75 @@ void Activation(Ref<Tensor> gate_, const Tensor& up, ActivationType type, cudaSt
     int num, dim;
     std::tie(num, dim) = gate.shapes(0, 1);
 
-    auto invoke = [&](auto t, auto act) {
-        using T = decltype(t);
+    auto invoke = [&](auto t, auto act, auto has_valid_tokens) {
+        using T                        = decltype(t);
+        constexpr bool kHasValidTokens = decltype(has_valid_tokens)::value;
 
         constexpr int vec_size = 4;
         constexpr int threads  = 512;
 
         const dim3 blocks(num, cdiv(dim, threads * vec_size));
-
-        ActivationKernel<vec_size><<<blocks, threads, 0, stream>>>(gate.data<T>(),  //
-                                                                   up.data<T>(),
-                                                                   act,
-                                                                   gate.stride(0),
-                                                                   num,
-                                                                   dim);
+        const int* valid_tokens = kHasValidTokens ? num_valid_tokens : nullptr;
+
+        ActivationKernel<kHasValidTokens, vec_size><<<blocks, threads, 0, stream>>>(gate.data<T>(),  //
+                                                                                    up.data<T>(),
+                                                                                    act,
+                                                                                    gate.stride(0),
+                                                                                    num,
+                                                                                    dim,
+                                                                                    valid_tokens);
     };
 
-    auto dispatch = [&](auto t) {
+    auto dispatch_act = [&](auto t, auto has_valid_tokens) {
         using T = decltype(t);
         if (type == ActivationType::kSilu) {
-            return invoke(t, Silu<T>{});
+            return invoke(t, Silu<T>{}, has_valid_tokens);
         }
         else if (type == ActivationType::kSiluGptOss) {
-            return invoke(t, SiluGptOss<T>{});
+            return invoke(t, SiluGptOss<T>{}, has_valid_tokens);
         }
         else {
             TM_LOG_FATAL("unknown activation type: {}", (int)type);
         }
     };
 
+    auto dispatch = [&](auto t) {
+        if (num_valid_tokens) {
+            return dispatch_act(t, std::true_type{});
+        }
+        return dispatch_act(t, std::false_type{});
+    };
+
     TM_DISPATCH_PRIMARY_DTYPES(gate.dtype(), dispatch);
     TM_CUDA_CHECK(cudaGetLastError());
 }
 
-template<int vec_size, class Activation, class T>
-__global__ void ActivationKernel(
-    T* gate_up, const T* bias, const int* group_ids, int64_t stride, Activation activation, int token_num, int dim)
+template<bool has_valid_tokens, int vec_size, class Activation, class T>
+__global__ void ActivationKernel(T*         gate_up,
+                                 const T*   bias,
+                                 const int* group_ids,
+                                 int64_t    stride,
+                                 Activation activation,
+                                 int        token_num,
+                                 int        dim,
+                                 const int* num_valid_tokens)
 {
     if constexpr (TURBOMIND_ARCH_DTYPE_GUARD(data_type_v<T>)) {
         const int di = (threadIdx.x + blockIdx.y * blockDim.x) * vec_size;
         const int ti = blockIdx.x;
-        const int gi = group_ids ? group_ids[ti] : 0;
+
+        if constexpr (has_valid_tokens) {
+            if (ti >= __ldg(num_valid_tokens)) {
+                return;
+            }
+        }
 
         if (di >= dim) {
             return;
         }
 
+        const int gi = group_ids ? group_ids[ti] : 0;
+
         using Vec = Array<T, vec_size>;
 
         Vec gate_bias{}, up_bias{};
@@ -145,6 +187,16 @@ void Activation(Tensor&             gate_up,  //
                 const Buffer_<int>& group_ids,
                 ActivationType      type,
                 cudaStream_t        stream)
+{
+    Activation(gate_up, bias, group_ids, type, nullptr, stream);
+}
+
+void Activation(Tensor&             gate_up,  //
+                const Tensor&       bias,
+                const Buffer_<int>& group_ids,
+                ActivationType      type,
+                const int*          num_valid_tokens,
+                cudaStream_t        stream)
 {
     const int num = gate_up.shape(0);
     const int dim = gate_up.shape(1) / 2;
@@ -153,42 +205,53 @@ void Activation(Tensor&             gate_up,  //
         Activation(gate_up.slice({0, 0}, {-1, dim}),  //
                    gate_up.slice({0, dim}, {-1, -1}),
                    type,
+                   num_valid_tokens,
                    stream);
         return;
     }
 
     TM_CHECK_EQ(gate_up.shape(-1), bias.shape(-1));
 
-    auto invoke = [&](auto t, auto act) {
-        using T = decltype(t);
+    auto invoke = [&](auto t, auto act, auto has_valid_tokens) {
+        using T                        = decltype(t);
+        constexpr bool kHasValidTokens = decltype(has_valid_tokens)::value;
 
         constexpr int vec_size = 4;
         constexpr int threads  = 512;
 
         const dim3 blocks(num, cdiv(dim, threads * vec_size));
-
-        ActivationKernel<vec_size><<<blocks, threads, 0, stream>>>(gate_up.data<T>(),  //
-                                                                   bias.data_or((T*)nullptr),
-                                                                   group_ids.data_or(nullptr),
-                                                                   gate_up.stride(0),
-                                                                   act,
-                                                                   num,
-                                                                   dim);
+        const int* valid_tokens = kHasValidTokens ? num_valid_tokens : nullptr;
+
+        ActivationKernel<kHasValidTokens, vec_size><<<blocks, threads, 0, stream>>>(gate_up.data<T>(),  //
+                                                                                    bias.data_or((T*)nullptr),
+                                                                                    group_ids.data_or(nullptr),
+                                                                                    gate_up.stride(0),
+                                                                                    act,
+                                                                                    num,
+                                                                                    dim,
+                                                                                    valid_tokens);
     };
 
-    auto dispatch = [&](auto t) {
+    auto dispatch_act = [&](auto t, auto has_valid_tokens) {
         using T = decltype(t);
         if (type == ActivationType::kSilu) {
-            return invoke(t, Silu<T>{});
+            return invoke(t, Silu<T>{}, has_valid_tokens);
         }
         else if (type == ActivationType::kSiluGptOss) {
-            return invoke(t, SiluGptOss<T>{});
+            return invoke(t, SiluGptOss<T>{}, has_valid_tokens);
         }
         else {
             TM_LOG_FATAL("unknown activation type: {}", (int)type);
         }
     };
 
+    auto dispatch = [&](auto t) {
+        if (num_valid_tokens) {
+            return dispatch_act(t, std::true_type{});
+        }
+        return dispatch_act(t, std::false_type{});
+    };
+
     TM_DISPATCH_PRIMARY_DTYPES(gate_up.dtype(), dispatch);
     TM_CUDA_CHECK(cudaGetLastError());
 }
diff --git a/src/turbomind/kernels/activation.h b/src/turbomind/kernels/activation.h
index 2eceeb9d89..72947b054c 100644
--- a/src/turbomind/kernels/activation.h
+++ b/src/turbomind/kernels/activation.h
@@ -14,10 +14,22 @@ enum class ActivationType
 
 void Activation(Ref<Tensor> gate, const Tensor& up, ActivationType type, cudaStream_t stream);
 
+// num_valid_tokens points to a device-side valid token count for padded token buffers.
+void Activation(
+    Ref<Tensor> gate, const Tensor& up, ActivationType type, const int* num_valid_tokens, cudaStream_t stream);
+
+void Activation(Tensor&             gate_up,  //
+                const Tensor&       bias,
+                const Buffer_<int>& group_ids,
+                ActivationType      type,
+                cudaStream_t        stream);
+
+// num_valid_tokens points to a device-side valid token count for padded token buffers.
 void Activation(Tensor&             gate_up,  //
                 const Tensor&       bias,
                 const Buffer_<int>& group_ids,
                 ActivationType      type,
+                const int*          num_valid_tokens,
                 cudaStream_t        stream);
 
 }  // namespace turbomind
diff --git a/src/turbomind/kernels/gemm/cublas.cu b/src/turbomind/kernels/gemm/cublas.cu
index f809500122..461c0d5036 100644
--- a/src/turbomind/kernels/gemm/cublas.cu
+++ b/src/turbomind/kernels/gemm/cublas.cu
@@ -310,9 +310,9 @@ public:
         const int N = Bdesc.cols;
         const int K = Adesc.cols;
 
-        if (ptr_offsets[group_count] != Adesc.rows) {
+        if (ptr_offsets[group_count] > Adesc.rows) {
             fprintf(stderr,
-                    "[TM][GEMM] CublasGrouped: offsets[%d]=%d != Adesc.rows=%d (would OOB)\n",
+                    "[TM][GEMM] CublasGrouped: offsets[%d]=%d exceeds Adesc.rows=%d (would OOB)\n",
                     group_count,
                     ptr_offsets[group_count],
                     Adesc.rows);
diff --git a/src/turbomind/kernels/gemm/moe_utils_v2.cu b/src/turbomind/kernels/gemm/moe_utils_v2.cu
index 2a6187cc5f..1045e1e2eb 100644
--- a/src/turbomind/kernels/gemm/moe_utils_v2.cu
+++ b/src/turbomind/kernels/gemm/moe_utils_v2.cu
@@ -572,17 +572,19 @@ template<int N>
 inline constexpr std::integral_constant<int, N> _Int{};
 
 void invokeMoeGate_V2(int*         f2n,            // [e*n] -> n
-                      int*         f2E,            // [e*n] -> E
+                      int*         f2E,            // [e*n] -> local E
                       int*         en2f,           // [e,n] -> n*e
-                      int*         offsets,        // [E+1]
+                      int*         offsets,        // [local E+1]
                       float*       scales,         // [e,n]
                       void*        masks,          // [E,n]
-                      int*         accum,          // [E]
-                      const float* logits,         // [e,n]
+                      int*         accum,          // [E,tiles]
+                      const float* logits,         // [n,E]
                       int          tokens,         //  n
                       int          tokens_padded,  //  round_up(n, 4)
                       int          experts,        //  E
                       int          experts_per_token,
+                      int          local_expert_offset,
+                      int          local_expert_num,
                       bool         softmax,
                       bool         norm_topk,
                       float        routed_scale,
@@ -671,19 +673,19 @@ void invokeMoeGate_V2(int*         f2n,            // [e*n] -> n
 
     {
         constexpr int threads = (1 << base_log_tile) / kMoeGateVecSize;
-        const dim3    blocks(tiles, experts + 1);
+        const dim3    blocks(tiles, local_expert_num + 1);
 
         MoeScanKernel_v2<threads><<<blocks, threads, 0, st>>>(f2n,  //
                                                               f2E,
                                                               en2f,
                                                               offsets,
-                                                              (int8_t*)masks,
-                                                              accum,
+                                                              (int8_t*)masks + local_expert_offset * tokens_padded,
+                                                              accum + local_expert_offset * tiles,
                                                               log_tile,
                                                               tiles,
                                                               tokens,
                                                               tokens_padded,
-                                                              experts);
+                                                              local_expert_num);
     }
     TM_CUDA_CHECK(cudaGetLastError());
 }
@@ -831,6 +833,8 @@ void invokeMoeGate_NoAuxTC(int*         f2n,
                            int          tokens_padded,
                            int          experts,
                            int          exp_per_tok,
+                           int          local_expert_offset,
+                           int          local_expert_num,
                            bool         norm_topk_prob,
                            float        routed_scale,
                            bool         use_sigmoid,
@@ -874,9 +878,19 @@ void invokeMoeGate_NoAuxTC(int*         f2n,
                                                           use_sigmoid);
 
     constexpr int scan_threads = (1 << base_log_tile) / kMoeGateVecSize;
-    const dim3    scan_blocks(tiles, experts + 1);
-    MoeScanKernel_v2<scan_threads><<<scan_blocks, scan_threads, 0, st>>>(
-        f2n, f2E, en2f, offsets, (int8_t*)masks, accum, log_tile, tiles, tokens, tokens_padded, experts);
+    const dim3    scan_blocks(tiles, local_expert_num + 1);
+    MoeScanKernel_v2<scan_threads>
+        <<<scan_blocks, scan_threads, 0, st>>>(f2n,
+                                               f2E,
+                                               en2f,
+                                               offsets,
+                                               (int8_t*)masks + local_expert_offset * tokens_padded,
+                                               accum + local_expert_offset * tiles,
+                                               log_tile,
+                                               tiles,
+                                               tokens,
+                                               tokens_padded,
+                                               local_expert_num);
 
     TM_CUDA_CHECK(cudaGetLastError());
 }
@@ -885,8 +899,13 @@ template<int vec_size, int block_dim, class T>
 __global__ void MoeGatherKernel(T*         dst,  // [e*n, d]
                                 const T*   src,  // [  n, d]
                                 const int* f2n,  // [e*n] :: e*n -> n
+                                const int* num_valid_tokens,
                                 int        dims)
 {
+    if (num_valid_tokens && blockIdx.x >= __ldg(num_valid_tokens)) {
+        return;
+    }
+
     using Vec        = Array<T, vec_size>;
     const int64_t bi = blockIdx.x;
 
@@ -899,23 +918,32 @@ __global__ void MoeGatherKernel(T*         dst,  // [e*n, d]
     }
 }
 
-void invokeMoeDispatch(Ref<Tensor> out_, const Tensor& src, const int* f2n, int expert_per_token, cudaStream_t st)
+void invokeMoeDispatch(Ref<Tensor>   out_,
+                       const Tensor& src,
+                       const int*    f2n,
+                       int           num_worst_tokens,
+                       const int*    num_valid_tokens,
+                       cudaStream_t  st)
 {
     auto& out    = out_.get();
     auto  invoke = [&](auto t) {
         using T                = decltype(t);
-        auto [num, dim]        = src.shapes(0, 1);
+        const int     dim      = src.shape(1);
         constexpr int threads  = 256;
         constexpr int vec_size = 16 / sizeof(T);
-        // std::cout << num * expert_per_token << " " << dim << "\n";
-        MoeGatherKernel<vec_size, threads><<<num * expert_per_token, threads, 0, st>>>(  //
+        // f2n/out have num_worst_tokens rows; num_valid_tokens limits rows that read f2n.
+        MoeGatherKernel<vec_size, threads><<<num_worst_tokens, threads, 0, st>>>(  //
             (T*)out.raw_data(),
             (const T*)src.raw_data(),
             f2n,
+            num_valid_tokens,
             dim / vec_size);
         TM_CUDA_CHECK(cudaGetLastError());
     };
     TM_CHECK_EQ(src.dtype(), out.dtype());
+    if (num_worst_tokens == 0) {
+        return;
+    }
     const auto elem_size = byte_size(src.dtype());
     if (elem_size == sizeof(uint16_t)) {
         invoke(uint16_t{});
@@ -958,10 +986,14 @@ __global__ void MoeDispatchScales(
 }
 
 template<class T>
-__global__ void
-MoeDispatchScalesNonaligned(T* dst, const T* src, int dst_stride, int src_stride, const int* f2n, int dim)
+__global__ void MoeDispatchScalesNonaligned(
+    T* dst, const T* src, int dst_stride, int src_stride, const int* f2n, const int* num_valid_tokens, int dim)
 {
     const int bi = blockIdx.x;
+    if (num_valid_tokens && bi >= __ldg(num_valid_tokens)) {
+        return;
+    }
+
     const int ti = f2n[bi];
 
     if (threadIdx.x < dim) {
@@ -969,14 +1001,20 @@ MoeDispatchScalesNonaligned(T* dst, const T* src, int dst_stride, int src_stride
     }
 }
 
-void invokeMoeDispatchScales(Ref<Tensor> out_, const Tensor& src, const int* f2n, int expert_per_token, cudaStream_t st)
+void invokeMoeDispatchScales(Ref<Tensor>   out_,
+                             const Tensor& src,
+                             const int*    f2n,
+                             int           num_worst_tokens,
+                             const int*    num_valid_tokens,
+                             cudaStream_t  st)
 {
     using T                 = float;
     constexpr int alignment = 16 / sizeof(T);
 
-    auto [dim, num] = src.shapes(0, 1);
+    const int dim = src.shape(0);
 
-    const int size         = num * expert_per_token;
+    // Keep the scale layout aligned to num_worst_tokens; num_valid_tokens limits rows that read f2n.
+    const int size         = num_worst_tokens;
     const int aligned_size = round_up<int>(size, alignment);
 
     auto& out = out_.get();
@@ -991,6 +1029,9 @@ void invokeMoeDispatchScales(Ref<Tensor> out_, const Tensor& src, const int* f2n
     }
 
     TM_CHECK_LE(dim, 1024);
+    if (size == 0) {
+        return;
+    }
     const int threads = round_up<int>(dim, WARP_SIZE);
     const int blocks  = size;
 
@@ -1001,6 +1042,7 @@ void invokeMoeDispatchScales(Ref<Tensor> out_, const Tensor& src, const int* f2n
                                                             out.stride(0),
                                                             src.stride(0),
                                                             f2n,
+                                                            num_valid_tokens,
                                                             dim);
 
     TM_CUDA_CHECK(cudaGetLastError());
@@ -1022,27 +1064,29 @@ __global__ void MoeReduceKernel(T*           dst,         // [  n, d]
     if constexpr (TURBOMIND_ARCH_DTYPE_GUARD(data_type_v<T>)) {
         const int64_t ti = blockIdx.x;
 
-        dst += dim * ti;
+        dst += (int64_t)dim * ti;
 
         if (dst_scales) {
-            dst_scale = dst_scales[ti];
-            dst_scale = fdividef(1.f, 1.f + expf(-dst_scale));
+            const float scale = dst_scales[ti];
+            dst_scale *= fdividef(1.f, 1.f + expf(-scale));
         }
 
         // Should be warp uniforms
-        const T* src_[exp_k];
-        const T* bias_[exp_k];
+        const T* src_[exp_k]{};
+        const T* bias_[exp_k]{};
 
-        float scale[exp_k];
+        float scale[exp_k]{};
 
         PRAGMA_UNROLL
         for (int e = 0; e < exp_k; ++e) {
             int fid = __ldg(&en2f[e * tokens + ti]);
-            src_[e] = src + dim * fid;
-            if constexpr (has_bias) {
-                bias_[e] = bias + __ldg(&f2E[fid]) * dim;
+            if (fid >= 0) {
+                src_[e] = src + (int64_t)dim * fid;
+                if constexpr (has_bias) {
+                    bias_[e] = bias + __ldg(&f2E[fid]) * (int64_t)dim;
+                }
+                scale[e] = scales ? __ldg(&scales[e * tokens + ti]) : 1.f;
             }
-            scale[e] = scales ? __ldg(&scales[e * tokens + ti]) : 1.f;
         }
 
         using Vec = Array<T, vec_size>;
@@ -1057,6 +1101,9 @@ __global__ void MoeReduceKernel(T*           dst,         // [  n, d]
             }
             PRAGMA_UNROLL
             for (int e = 0; e < exp_k; ++e) {
+                if (src_[e] == nullptr) {
+                    continue;
+                }
                 Vec v;
                 Load(v, src_[e] + i);
                 using namespace ops;
diff --git a/src/turbomind/kernels/gemm/moe_utils_v2.h b/src/turbomind/kernels/gemm/moe_utils_v2.h
index ae72148c95..ca0053fddc 100644
--- a/src/turbomind/kernels/gemm/moe_utils_v2.h
+++ b/src/turbomind/kernels/gemm/moe_utils_v2.h
@@ -26,21 +26,29 @@ void invokeMoeGate_V2(int*         f2n,
                       int          tokens_padded,
                       int          experts,
                       int          exp_per_tok,
+                      int          local_expert_offset,
+                      int          local_expert_num,
                       bool         softmax,
                       bool         norm_topk,
                       float        routed_scale,
                       cudaStream_t st);
 
+// num_worst_tokens is the output capacity / launch upper bound for f2n/out.
+// If num_valid_tokens is set, it points to a device-side count of valid rows and
+// rows >= *num_valid_tokens return before reading f2n.
 void invokeMoeDispatch(Ref<Tensor>   out_,  //
                        const Tensor& src,
                        const int*    f2n,
-                       int           expert_per_token,
+                       int           num_worst_tokens,
+                       const int*    num_valid_tokens,
                        cudaStream_t  st);
 
+// Same num_worst_tokens / num_valid_tokens contract as invokeMoeDispatch
 void invokeMoeDispatchScales(Ref<Tensor>   out_,  //
                              const Tensor& src,
                              const int*    f2n,
-                             int           expert_per_token,
+                             int           num_worst_tokens,
+                             const int*    num_valid_tokens,
                              cudaStream_t  st);
 
 void invokeMoeCombine(Ref<Tensor>   out_,
@@ -75,6 +83,8 @@ void invokeMoeGate_NoAuxTC(int*         f2n,
                            int          tokens_padded,
                            int          experts,
                            int          exp_per_tok,
+                           int          local_expert_offset,
+                           int          local_expert_num,
                            bool         norm_topk_prob,
                            float        routed_scale,
                            bool         use_sigmoid,
diff --git a/src/turbomind/kernels/gemm/test/test_moe_utils.cu b/src/turbomind/kernels/gemm/test/test_moe_utils.cu
index f5338cc4ad..37e7b79a85 100644
--- a/src/turbomind/kernels/gemm/test/test_moe_utils.cu
+++ b/src/turbomind/kernels/gemm/test/test_moe_utils.cu
@@ -233,6 +233,8 @@ bool test_moe_gate(int                     tokens,  //
                          tokens_padded,
                          expert_num,
                          experts_per_token,
+                         0,
+                         expert_num,
                          softmax,
                          false,
                          1.f,
diff --git a/src/turbomind/kernels/gemm/test/testbed_v3.h b/src/turbomind/kernels/gemm/test/testbed_v3.h
index 805b9aca02..d633baa18b 100644
--- a/src/turbomind/kernels/gemm/test/testbed_v3.h
+++ b/src/turbomind/kernels/gemm/test/testbed_v3.h
@@ -378,7 +378,7 @@ struct Testbed_v3: Parameter {
         Tensor xe{{x.shape(0) * experts_per_token, input_dim}, data_type, kDEVICE};
         Tensor de{{x.shape(0) * experts_per_token, output_dim}, data_type, kDEVICE};
 
-        TM_SCOPE_CALL(invokeMoeDispatch(xe, x, f2n_.data(), experts_per_token, stream_));
+        TM_SCOPE_CALL(invokeMoeDispatch(xe, x, f2n_.data(), x.shape(0) * experts_per_token, nullptr, stream_));
 
         for (int i = 0; i < expert_num; ++i) {
             const int base = h_offsets_[i], size = h_offsets_[i + 1] - base;
diff --git a/src/turbomind/models/llama/LlamaLinear.cu b/src/turbomind/models/llama/LlamaLinear.cu
index 515a1eb8dd..f0a3126c7e 100644
--- a/src/turbomind/models/llama/LlamaLinear.cu
+++ b/src/turbomind/models/llama/LlamaLinear.cu
@@ -84,13 +84,13 @@ struct LlamaLinear::Impl {
         // SM100+ grouped bf16/fp16: use chunk() weights so Activation() runs separately.
         const bool is_cublas_grouped = offsets && getSMVersion() == 100 && weight.weight_format.dtype == kBfloat16;
         if (indices && (A.dtype() == kFloat8_e4m3 || is_cublas_grouped)) {
-            const auto [bsz, k] = A.shapes(0, 1);
-            const int e         = indices.size() / bsz;
-            Tensor    A_e       = {{m, k}, A.dtype(), kDEVICE};
-            TM_SCOPE_CALL(invokeMoeDispatch(A_e, A, indices.data(), e, st));
+            const int  k                = A.shape(1);
+            Tensor     A_e              = {{m, k}, A.dtype(), kDEVICE};
+            const int* num_valid_tokens = offsets ? offsets.data() + offsets.size() - 1 : nullptr;
+            TM_SCOPE_CALL(invokeMoeDispatch(A_e, A, indices.data(), m, num_valid_tokens, st));
             if (U) {
                 Tensor U_e;
-                TM_SCOPE_CALL(invokeMoeDispatchScales(U_e, U, indices.data(), e, st));
+                TM_SCOPE_CALL(invokeMoeDispatchScales(U_e, U, indices.data(), m, num_valid_tokens, st));
                 U = U_e;
             }
             A       = A_e;
diff --git a/src/turbomind/models/llama/llama_params.h b/src/turbomind/models/llama/llama_params.h
index 4ce0a586fa..23cc89d82a 100644
--- a/src/turbomind/models/llama/llama_params.h
+++ b/src/turbomind/models/llama/llama_params.h
@@ -15,6 +15,7 @@ struct EngineParam: EngineConfig {
     int attn_tp_rank  = 0;
     int attn_cp_rank  = 0;
     int mlp_tp_rank   = 0;
+    int ep_rank       = 0;
     int model_tp_rank = 0;  // rank(d_tp_group), in [0, attn_tp_size × attn_cp_size)
 
     // Derived field (set in Impl ctor)
diff --git a/src/turbomind/models/llama/moe_ffn_layer.cc b/src/turbomind/models/llama/moe_ffn_layer.cc
index ffff786e66..18a44874f2 100644
--- a/src/turbomind/models/llama/moe_ffn_layer.cc
+++ b/src/turbomind/models/llama/moe_ffn_layer.cc
@@ -1,10 +1,13 @@
 // Copyright (c) OpenMMLab. All rights reserved.
 
+#include <memory>
+
 #include <cuda_runtime.h>
 
 #include "src/turbomind/core/context.h"
 #include "src/turbomind/core/scope.h"
 #include "src/turbomind/kernels/activation.h"
+#include "src/turbomind/kernels/gemm/moe_utils_v2.h"
 #include "src/turbomind/kernels/norm/rms_norm.h"
 
 #include "src/turbomind/models/llama/LlamaLinear.h"
@@ -19,21 +22,101 @@
 
 namespace turbomind {
 
-MoeFfnLayer::MoeFfnLayer(const EngineParam& engine, const Context& ctx):
+class MoeFfnLayerImpl {
+public:
+    explicit MoeFfnLayerImpl(const Context& ctx): linear_(*ctx.linear) {}
+
+    virtual ~MoeFfnLayerImpl() = default;
+
+    virtual void Forward(MoeFfnLayer::ForwardParam& p) = 0;
+
+    virtual void Combine(MoeFfnLayer::ForwardParam& p) = 0;
+
+    virtual Tensor GetShardFfnInput(Tensor& global_hidden_states) = 0;
+
+protected:
+    Tensor_<float> Gate(const Tensor& input, const LinearWeight& gate);
+
+    LlamaLinear& linear_;
+};
+
+Tensor_<float> MoeFfnLayerImpl::Gate(const Tensor& input, const LinearWeight& gate)
+{
+    TM_FUNCTION_SCOPE();
+
+    auto& w = gate.weight;
+    TM_CHECK_EQ(input.shape(1), w.shape(0));
+    Tensor_<float> logits{{input.shape(0), w.shape(1)}, kDEVICE};
+    TM_SCOPE_CALL(linear_.Forward(input, gate, logits));
+    ApplyBias(logits, gate.bias, core::Context::stream().handle());
+    TM_CUDA_CHECK(cudaGetLastError());
+    return logits;
+}
+
+class MoeFfnDefaultImpl final: public MoeFfnLayerImpl {
+public:
+    MoeFfnDefaultImpl(const EngineParam& engine, const Context& ctx);
+
+    void Forward(MoeFfnLayer::ForwardParam& p) override;
+
+    void Combine(MoeFfnLayer::ForwardParam& p) override;
+
+    Tensor GetShardFfnInput(Tensor& global_hidden_states) override;
+
+private:
+    void Init(MoeFfnLayer::ForwardParam& p);
+
+    const int tp_size_;
+    const int ep_size_;
+    const int ep_rank_;
+    const int max_token_num_;
+    int&      is_warm_up_;
+
+    bool initialized_ = false;
+
+    Buffer_<int> h_offsets_;
+
+    Buffer_<int>   masks_;
+    Buffer_<int>   f2n_;
+    Buffer_<int>   f2E_;
+    Buffer_<int>   en2f_;
+    Buffer_<float> scales_;
+    Buffer_<int>   accum_;
+    Buffer_<int>   offsets_;
+
+    Tensor         temp_;
+    Tensor_<float> shared_scales_;
+
+    // When a MOE model enables EP inference and the model includes dense layers or shared experts, an additional stream
+    // is used to perform sharding and cleaning of the FFN inputs.
+    Stream clear_stream_;
+    Event  clear_ready_event_;
+    Event  clear_done_event_;
+    bool   clear_pending_ = false;
+};
+
+MoeFfnDefaultImpl::MoeFfnDefaultImpl(const EngineParam& engine, const Context& ctx):
+    MoeFfnLayerImpl(ctx),
     tp_size_(engine.mlp_tp_size),
+    ep_size_(engine.ep_size),
+    ep_rank_(engine.ep_rank),
     max_token_num_(engine.max_forward_token_num * engine.attn_dp_size),
-    is_warm_up_(*ctx.is_warm_up),
-    linear_(*ctx.linear),
-    expert_ffn_(std::make_unique<LlamaFfnLayer>(ctx))
+    is_warm_up_(*ctx.is_warm_up)
 {
+    if (ep_size_ > 1) {
+        clear_stream_      = Stream::create();
+        clear_ready_event_ = Event::create();
+        clear_done_event_  = Event::create();
+    }
 }
 
-void MoeFfnLayer::Init(ForwardParam& p)
+void MoeFfnDefaultImpl::Init(MoeFfnLayer::ForwardParam& p)
 {
     const int expert_num        = p.weights->num_experts();
+    const int local_expert_num  = p.weights->num_local_experts();
     const int experts_per_token = p.weights->experts_per_token;
 
-    h_offsets_ = {expert_num + 1, kCPU};
+    h_offsets_ = {local_expert_num + 1, kCPU};
 
     const int pad_token_num = (max_token_num_ + kMoeGateVecSize - 1) / kMoeGateVecSize * kMoeGateVecSize;
 
@@ -42,26 +125,13 @@ void MoeFfnLayer::Init(ForwardParam& p)
     f2E_     = {experts_per_token * max_token_num_, kDEVICE};
     en2f_    = {experts_per_token * max_token_num_, kDEVICE};
     scales_  = {experts_per_token * max_token_num_, kDEVICE};
-    offsets_ = {expert_num + 1, kDEVICE};
+    offsets_ = {local_expert_num + 1, kDEVICE};
     accum_   = {expert_num * kMoeGateMaxTiles, kDEVICE};
 
     initialized_ = true;
 }
 
-Tensor_<float> MoeFfnLayer::Gate(const Tensor& input, const LinearWeight& gate)
-{
-    TM_FUNCTION_SCOPE();
-
-    auto& w = gate.weight;
-    TM_CHECK_EQ(input.shape(1), w.shape(0));
-    Tensor_<float> logits{{input.shape(0), w.shape(1)}, kDEVICE};
-    TM_SCOPE_CALL(linear_.Forward(input, gate, logits));
-    ApplyBias(logits, gate.bias, core::Context::stream().handle());
-    TM_CUDA_CHECK(cudaGetLastError());
-    return logits;
-}
-
-void MoeFfnLayer::Forward(ForwardParam& p)
+void MoeFfnDefaultImpl::Forward(MoeFfnLayer::ForwardParam& p)
 {
     TM_FUNCTION_SCOPE();
     if (!initialized_) {
@@ -76,8 +146,12 @@ void MoeFfnLayer::Forward(ForwardParam& p)
     const int hidden_dim = block.hidden_dim;
     const int inter_size = block.inter_size;
 
-    const size_t padded     = (tokens + kMoeGateVecSize - 1) / kMoeGateVecSize * kMoeGateVecSize;
-    const int    expert_num = moe.num_experts();
+    const size_t padded = (tokens + kMoeGateVecSize - 1) / kMoeGateVecSize * kMoeGateVecSize;
+
+    const int expert_num        = moe.num_experts();
+    const int local_expert_num  = moe.num_local_experts();
+    const int expert_offset     = moe.local_expert_offset();
+    const int experts_per_token = moe.experts_per_token;
 
     TM_CHECK(expert_num);
 
@@ -87,8 +161,11 @@ void MoeFfnLayer::Forward(ForwardParam& p)
 
     const auto st = core::Context::stream().handle();
 
+    if (ep_size_ > 1) {
+        TM_CUDA_CHECK(cudaMemsetAsync(en2f_.data(), -1, sizeof(int) * tokens * experts_per_token, st));
+    }
+
     if (p.weights->topk_method == "noaux_tc") {
-        // invokeMoeGate_NoAuxTC clears accum and masks internally
         TM_CHECK_EQ(p.weights->n_group, 1);
         TM_CHECK_EQ(p.weights->topk_group, 1);
         const float* correction_bias = nullptr;
@@ -107,7 +184,9 @@ void MoeFfnLayer::Forward(ForwardParam& p)
                               tokens,
                               padded,
                               expert_num,
-                              p.weights->experts_per_token,
+                              experts_per_token,
+                              expert_offset,
+                              local_expert_num,
                               p.weights->norm_topk_prob,
                               p.weights->routed_scale,
                               p.weights->scoring_func == "sigmoid",
@@ -137,7 +216,9 @@ void MoeFfnLayer::Forward(ForwardParam& p)
                          tokens,
                          padded,
                          expert_num,
-                         p.weights->experts_per_token,
+                         experts_per_token,
+                         expert_offset,
+                         local_expert_num,
                          softmax,
                          p.weights->norm_topk_prob,
                          p.weights->routed_scale,
@@ -147,31 +228,40 @@ void MoeFfnLayer::Forward(ForwardParam& p)
 
     if (is_warm_up_) {
         std::mt19937     g;
-        const auto       expert_ids = SampleUniform(tokens, expert_num, p.weights->experts_per_token, g);
-        std::vector<int> cnt(expert_num);
+        const auto       expert_ids = SampleUniform(tokens, local_expert_num, experts_per_token, g);
+        std::vector<int> cnt(local_expert_num);
         for (const auto& x : expert_ids) {
             ++cnt[x];
         }
         h_offsets_[0] = 0;
-        for (int i = 0; i < expert_num; ++i) {
+        for (int i = 0; i < local_expert_num; ++i) {
             h_offsets_[i + 1] = h_offsets_[i] + cnt[i];
         }
-        TM_CUDA_CHECK(
-            cudaMemcpyAsync(offsets_.data(), h_offsets_.data(), sizeof(int) * (expert_num + 1), cudaMemcpyDefault, st));
+        TM_CUDA_CHECK(cudaMemcpyAsync(
+            offsets_.data(), h_offsets_.data(), sizeof(int) * (local_expert_num + 1), cudaMemcpyDefault, st));
+
+        if (ep_size_ > 1) {
+            const auto entries = static_cast<size_t>(tokens) * experts_per_token;
+            TM_CUDA_CHECK(cudaMemsetAsync(f2n_.data(), 0, sizeof(int) * entries, st));
+            TM_CUDA_CHECK(cudaMemsetAsync(f2E_.data(), 0, sizeof(int) * entries, st));
+            TM_CUDA_CHECK(cudaMemsetAsync(en2f_.data(), -1, sizeof(int) * entries, st));
+        }
     }
 
-    temp_ = Tensor{{p.weights->experts_per_token * tokens, hidden_dim}, p.input.dtype(), p.input.device()};
+    temp_ = Tensor{{tokens * experts_per_token, hidden_dim}, p.input.dtype(), p.input.device()};
+
+    // For ep_size > 1, the valid tokens are less than tokens * experts_per_token
+    const int* num_valid_tokens = ep_size_ > 1 ? offsets_.data() + local_expert_num : nullptr;
 
-    auto indices = f2n_.slice(0, tokens * p.weights->experts_per_token);
-    auto offsets = offsets_.slice(0, expert_num + 1);
+    auto indices = f2n_.slice(0, temp_.shape(0));
+    auto offsets = offsets_.slice(0, local_expert_num + 1);
 
     if (block.w1w3) {
-        // Fused w1w3 path
         Tensor inter;
-        TM_SCOPE_CALL(linear_.Forward(p.input, *block.w1w3, indices, offsets_, inter));
+        TM_SCOPE_CALL(linear_.Forward(p.input, *block.w1w3, indices, offsets, inter));
 
         if (!block.is_fused_silu) {
-            Activation(inter, block.w1w3->bias, f2E_, block.act_type, st);
+            Activation(inter, block.w1w3->bias, f2E_, block.act_type, num_valid_tokens, st);
             TM_CUDA_CHECK(cudaGetLastError());
         }
 
@@ -180,12 +270,12 @@ void MoeFfnLayer::Forward(ForwardParam& p)
     else {
         // Separate w1/w3 path
         Tensor gating;
-        TM_SCOPE_CALL(linear_.Forward(p.input, *block.w1, indices, offsets_, gating));
+        TM_SCOPE_CALL(linear_.Forward(p.input, *block.w1, indices, offsets, gating));
 
         Tensor up;
-        TM_SCOPE_CALL(linear_.Forward(p.input, *block.w3, indices, offsets_, up));
+        TM_SCOPE_CALL(linear_.Forward(p.input, *block.w3, indices, offsets, up));
 
-        Activation(gating, up, block.act_type, st);
+        Activation(gating, up, block.act_type, num_valid_tokens, st);
         TM_CUDA_CHECK(cudaGetLastError());
 
         TM_SCOPE_CALL(linear_.Forward(gating, *block.w2, {}, offsets, temp_));
@@ -196,19 +286,24 @@ void MoeFfnLayer::Forward(ForwardParam& p)
     }
 }
 
-void MoeFfnLayer::Combine(ForwardParam& p)
+void MoeFfnDefaultImpl::Combine(MoeFfnLayer::ForwardParam& p)
 {
     TM_FUNCTION_SCOPE();
     auto& moe = *p.weights;
 
+    if (clear_pending_) {
+        core::Context::stream().Wait(clear_done_event_);
+        clear_pending_ = false;
+    }
+
     invokeMoeCombine(p.output,
                      temp_,
-                     moe.block()->w2->bias,
+                     TM_CHECK_NOTNULL(moe.block())->w2->bias,
                      scales_.data(),
                      en2f_.data(),
                      f2E_.data(),
                      shared_scales_.data_or((float*)nullptr),
-                     p.weights->experts_per_token,
+                     moe.experts_per_token,
                      1.f / tp_size_,
                      p.scale,
                      core::Context::stream().handle());
@@ -218,4 +313,66 @@ void MoeFfnLayer::Combine(ForwardParam& p)
     shared_scales_ = {};
 }
 
+Tensor MoeFfnDefaultImpl::GetShardFfnInput(Tensor& global_hidden_states)
+{
+    TM_FUNCTION_SCOPE();
+    if (ep_size_ == 1) {
+        return global_hidden_states;
+    }
+
+    TM_CHECK(!clear_pending_);
+    const int token_num         = global_hidden_states.shape(0);
+    const int tokens_per_rank   = token_num / ep_size_;
+    const int remainder         = token_num % ep_size_;
+    const int local_token_num   = tokens_per_rank + (ep_rank_ < remainder);
+    const int local_token_begin = ep_rank_ * tokens_per_rank + std::min(ep_rank_, remainder);
+    const int local_token_end   = local_token_begin + local_token_num;
+
+    if (local_token_begin > 0 || local_token_end < token_num) {
+        auto& stream = core::Context::stream();
+
+        clear_ready_event_.Record(stream);
+        clear_stream_.Wait(clear_ready_event_);
+
+        if (local_token_begin > 0) {
+            Clear(global_hidden_states.slice(0, local_token_begin), clear_stream_);
+        }
+        if (local_token_end < token_num) {
+            Clear(global_hidden_states.slice(local_token_end, token_num - local_token_end), clear_stream_);
+        }
+
+        clear_done_event_.Record(clear_stream_);
+        clear_pending_ = true;
+    }
+
+    return global_hidden_states.slice(local_token_begin, local_token_num);
+}
+
+MoeFfnLayer::MoeFfnLayer(const EngineParam& engine, const Context& ctx)
+{
+    if (engine.ep_size <= 1 || engine.nnodes == 1) {
+        impl_ = std::make_unique<MoeFfnDefaultImpl>(engine, ctx);
+        return;
+    }
+
+    TM_LOG_FATAL("Unsupported config for MoeFfnLayer");
+}
+
+MoeFfnLayer::~MoeFfnLayer() = default;
+
+void MoeFfnLayer::Forward(ForwardParam& p)
+{
+    impl_->Forward(p);
+}
+
+void MoeFfnLayer::Combine(ForwardParam& p)
+{
+    impl_->Combine(p);
+}
+
+Tensor MoeFfnLayer::GetShardFfnInput(Tensor& global_hidden_states)
+{
+    return impl_->GetShardFfnInput(global_hidden_states);
+}
+
 }  // namespace turbomind
diff --git a/src/turbomind/models/llama/moe_ffn_layer.h b/src/turbomind/models/llama/moe_ffn_layer.h
index d50bc4869b..ea366440a6 100644
--- a/src/turbomind/models/llama/moe_ffn_layer.h
+++ b/src/turbomind/models/llama/moe_ffn_layer.h
@@ -2,18 +2,23 @@
 
 #pragma once
 
-#include "src/turbomind/kernels/gemm/context.h"
-#include "src/turbomind/kernels/gemm/moe_utils_v2.h"
-#include "src/turbomind/models/llama/LlamaFfnLayer.h"
+#include <memory>
+
+#include "src/turbomind/core/core.h"
+#include "src/turbomind/models/llama/context.h"
 #include "src/turbomind/models/llama/llama_params.h"
 #include "src/turbomind/models/moe_weight.h"
 
 namespace turbomind {
 
+class MoeFfnLayerImpl;
+
 class MoeFfnLayer {
 public:
     MoeFfnLayer(const EngineParam& engine, const Context& ctx);
 
+    ~MoeFfnLayer();
+
     struct ForwardParam {
         Tensor           input;
         Tensor           output;
@@ -26,38 +31,10 @@ class MoeFfnLayer {
 
     void Combine(ForwardParam& p);
 
-private:
-    void Init(ForwardParam& p);
-
-    Tensor_<float> Gate(const Tensor& input, const LinearWeight& gate);
-
-    void dump_logits(int token_num, int layer_id, int expert_num);
-
-    const int tp_size_;
-    const int max_token_num_;
-    int&      is_warm_up_;
+    Tensor GetShardFfnInput(Tensor& global_hidden_states);
 
-    LlamaLinear& linear_;
-
-    std::unique_ptr<LlamaFfnLayer> expert_ffn_;
-
-    bool initialized_ = false;
-
-    ///////////////////////////////////////////////////////
-    /// runtime states
-    Buffer_<int> h_offsets_;
-
-    Buffer_<int>   masks_;
-    Buffer_<int>   f2n_;
-    Buffer_<int>   f2E_;
-    Buffer_<int>   en2f_;
-    Buffer_<float> scales_;
-    Buffer_<int>   accum_;
-    Buffer_<int>   offsets_;
-
-    Tensor         temp_;
-    Tensor_<float> shared_scales_;
-    ///////////////////////////////////////////////////////
+private:
+    std::unique_ptr<MoeFfnLayerImpl> impl_;
 };
 
 }  // namespace turbomind
diff --git a/src/turbomind/models/llama/unified_decoder.cc b/src/turbomind/models/llama/unified_decoder.cc
index ba7857c7c6..ba0fbfe743 100644
--- a/src/turbomind/models/llama/unified_decoder.cc
+++ b/src/turbomind/models/llama/unified_decoder.cc
@@ -96,6 +96,8 @@ UnifiedDecoder::UnifiedDecoder(CacheRegistry&     registry,
                                                                   ctx,
                                                                   phases);
     }
+
+    TM_CHECK(!(moe_weights.empty() && engine.ep_size > 1)) << "Dense model is not supported with ep_size > 1";
 }
 
 void UnifiedDecoder::AllreduceResidualRMSnorm(Tensor&       hidden_states,
@@ -289,14 +291,18 @@ void UnifiedDecoder::Forward(int phase, TensorMap& args, const std::vector<Weigh
             moe_fwd_param = MoeFfnLayer::ForwardParam{global_hidden_states,
                                                       global_hidden_states,
                                                       weights.at(layer)->moe_ffn.get(),
-                                                      ffn_layer_ ? 1.f : 0.f,
+                                                      weights.at(layer)->feed_forward ? 1.f : 0.f,
                                                       layer};
             moe_ffn_layer_->Forward(*moe_fwd_param);
         }
 
         if (ffn_layer_ && weights.at(layer)->feed_forward) {
-            ffn_layer_->forward(
-                {global_hidden_states, global_hidden_states, weights.at(layer)->feed_forward.get(), (int)layer});
+            auto ffn_input_shared =
+                moe_ffn_layer_ ? moe_ffn_layer_->GetShardFfnInput(global_hidden_states) : global_hidden_states;
+            if (ffn_input_shared.shape(0) > 0) {
+                ffn_layer_->forward(
+                    {ffn_input_shared, ffn_input_shared, weights.at(layer)->feed_forward.get(), (int)layer});
+            }
         }
 
         if (moe_fwd_param) {
diff --git a/src/turbomind/models/moe_weight.cc b/src/turbomind/models/moe_weight.cc
index 2020e82e1c..3c34b2a183 100644
--- a/src/turbomind/models/moe_weight.cc
+++ b/src/turbomind/models/moe_weight.cc
@@ -23,6 +23,8 @@ MoeWeight::MoeWeight(const core::MoeConfig& cfg)
     act_type_         = static_cast<ActivationType>(cfg.act_type);
     fuse_silu_act_    = cfg.fuse_silu;
     expert_num        = cfg.expert_num;
+    ep_size           = cfg.ep_size;
+    ep_rank           = cfg.ep_rank;
 }
 
 // Adapted from LinkExperts for LinearWeight
@@ -79,7 +81,7 @@ FfnWeight* MoeWeight::expert(int i) const
     if (!experts) {
         return nullptr;
     }
-    return static_cast<FfnWeight*>(experts->child(std::to_string(i)));
+    return static_cast<FfnWeight*>(experts->child(std::to_string(local_expert_offset() + i)));
 }
 
 void MoeWeight::prepare()
@@ -87,6 +89,8 @@ void MoeWeight::prepare()
     // First prepare all children (experts, gate, etc.)
     Module::prepare();
 
+    const int local_expert_num = num_local_experts();
+
     // Create batched block view for fused MoE path
     auto e0 = TM_CHECK_NOTNULL(expert(0));  // exemplar expert
 
@@ -123,23 +127,23 @@ void MoeWeight::prepare()
     if (get_expert_w1w3(0)) {
         // Fused w1w3 path: experts have a single fused gate+up projection
         block_->add_child("w1w3", std::make_unique<LinearWeight>());
-        LinkLinearExperts(get_expert_w1w3, expert_num, *block_->w1w3);
+        LinkLinearExperts(get_expert_w1w3, local_expert_num, *block_->w1w3);
     }
     else {
         // Separate w1/w3 path: link individually
         block_->add_child("w1", std::make_unique<LinearWeight>());
         block_->add_child("w3", std::make_unique<LinearWeight>());
         if (get_expert_w1(0)) {
-            LinkLinearExperts(get_expert_w1, expert_num, *block_->w1);
+            LinkLinearExperts(get_expert_w1, local_expert_num, *block_->w1);
         }
         if (get_expert_w3(0)) {
-            LinkLinearExperts(get_expert_w3, expert_num, *block_->w3);
+            LinkLinearExperts(get_expert_w3, local_expert_num, *block_->w3);
         }
     }
 
     block_->add_child("w2", std::make_unique<LinearWeight>());
     if (get_expert_w2(0)) {
-        LinkLinearExperts(get_expert_w2, expert_num, *block_->w2);
+        LinkLinearExperts(get_expert_w2, local_expert_num, *block_->w2);
     }
 
     // Propagate the actual fused-silu state from the first expert to
diff --git a/src/turbomind/models/moe_weight.h b/src/turbomind/models/moe_weight.h
index 4b767e6710..9bdeea5d93 100644
--- a/src/turbomind/models/moe_weight.h
+++ b/src/turbomind/models/moe_weight.h
@@ -26,6 +26,8 @@ struct MoeConfig: ModuleConfig {
     X(int, n_group)                                                                                                    \
     X(int, router_n_groups)                                                                                            \
     X(double, routed_scale)                                                                                            \
+    X(int, ep_size, 1)                                                                                                 \
+    X(int, ep_rank, 0)                                                                                                 \
     X(DataType, data_type)
 
     MOE_FIELDS(TM_MEMBER)
@@ -54,6 +56,14 @@ class MoeWeight: public core::Module {
     {
         return expert_num;
     }
+    int num_local_experts() const
+    {
+        return expert_num / ep_size;
+    }
+    int local_expert_offset() const
+    {
+        return ep_rank * num_local_experts();
+    }
 
     // --- X-macro child members ---
 #define MOE_WEIGHT_CHILDREN(X)                                                                                         \
@@ -82,6 +92,8 @@ class MoeWeight: public core::Module {
     int         n_group{};
     std::string scoring_func;
     int         router_n_groups{};
+    int         ep_size{1};
+    int         ep_rank{0};
 
 private:
     ActivationType act_type_{};
diff --git a/src/turbomind/python/bind.cpp b/src/turbomind/python/bind.cpp
index 0593337b08..1e2f9b0603 100644
--- a/src/turbomind/python/bind.cpp
+++ b/src/turbomind/python/bind.cpp
@@ -803,5 +803,6 @@ PYBIND11_MODULE(_turbomind, m)
         .def("is_dummy_node", [](TurboMind* model) { return model->is_dummy_node(); })
         .def("attn_tp_rank", &TurboMind::GetAttnTpRank, "index"_a)
         .def("mlp_tp_rank", &TurboMind::GetMlpTpRank, "index"_a)
+        .def("ep_rank", &TurboMind::GetEpRank, "index"_a)
         .def("model_tp_rank", &TurboMind::GetModelTpRank, "index"_a);
 }
diff --git a/src/turbomind/turbomind.cc b/src/turbomind/turbomind.cc
index e4c40bb79b..1a9ff0e0dc 100644
--- a/src/turbomind/turbomind.cc
+++ b/src/turbomind/turbomind.cc
@@ -168,7 +168,7 @@ TurboMind::Impl::Impl(string model_dir, EngineConfig config, FFICtxFactory ffi_c
     }
 
     comm_size_ = engine_param_.attn_dp_size * engine_param_.attn_tp_size * engine_param_.attn_cp_size;
-    TM_CHECK(engine_param_.mlp_tp_size == comm_size_);
+    TM_CHECK(engine_param_.mlp_tp_size * engine_param_.ep_size == comm_size_);
 
     communicator_type_ = std::move(config.communicator);
 
@@ -244,7 +244,8 @@ void TurboMind::Impl::CreateContext(int index)
 
         p.model_tp_rank = c.d_comm->rank(c.d_tp_group);
         p.attn_tp_rank  = p.model_tp_rank / p.attn_cp_size;
-        p.mlp_tp_rank   = c.d_comm->rank(0);
+        p.mlp_tp_rank   = inner_rank % p.mlp_tp_size;
+        p.ep_rank       = inner_rank / p.mlp_tp_size;
     }
 
     if (c.h_tp_group->rank() == 0) {
@@ -540,6 +541,11 @@ int TurboMind::GetMlpTpRank(int index)
     return impl_->engine_params_.at(index).mlp_tp_rank;
 }
 
+int TurboMind::GetEpRank(int index)
+{
+    return impl_->engine_params_.at(index).ep_rank;
+}
+
 int TurboMind::GetModelTpRank(int index)
 {
     return impl_->engine_params_.at(index).model_tp_rank;
diff --git a/src/turbomind/turbomind.h b/src/turbomind/turbomind.h
index 4d19d12641..74d602493c 100644
--- a/src/turbomind/turbomind.h
+++ b/src/turbomind/turbomind.h
@@ -52,6 +52,9 @@ class TurboMind {
     /// MLP TP rank for GPU *index*.
     int GetMlpTpRank(int index);
 
+    /// Expert-parallel rank for GPU *index*.
+    int GetEpRank(int index);
+
     /// Model-level TP rank (rank within d_tp_group) for GPU *index*.
     int GetModelTpRank(int index);