code: def create_ep_hierarchical_group(
      ep_group_ranks: List[int],
      nproc_per_node: Optional[int] = None,
  ) -> Tuple[int, dist.ProcessGroup, Optional[dist.ProcessGroup]]:
      """
      e.g., If ep_group = [1, 2, 5, 6], and nproc_per_node = 4
          Then, ep_intra_group = [1, 2] & [5, 6], ep_inter_group = [1, 5] & None
      """
      assert (
          dist.is_initialized()
      ), "Please initialize torch.distributed first."
      rank = dist.get_rank()
      if nproc_per_node is None:
          nproc_per_node = os.environ.get("LOCAL_WORLD_SIZE")
          assert (
              nproc_per_node is not None
          ), "Please use torchrun to launch the job, or specify nproc_per_node manually."
          nproc_per_node = int(nproc_per_node)
      else:
          assert (
              dist.get_world_size() % nproc_per_node == 0
          ), "nproc_per_node should be a divisor of world_size."
      num_node = dist.get_world_size() // nproc_per_node
  
      intra_src_rank = None
      ep_intra_node_group = None
      for i in range(num_node):
          ep_intra_ranks = [
              i * nproc_per_node + j
              for j in range(nproc_per_node)
              if j in ep_group_ranks
          ]
          group = dist.new_group(ep_intra_ranks)
          if rank in ep_intra_ranks:
              assert ep_intra_node_group is None
              ep_intra_node_group = group
              intra_src_rank = ep_intra_ranks[0]
  
      ep_inter_node_group = None
      ep_inter_ranks = [
          ep_group_ranks[0] + i * nproc_per_node
          for i in range(num_node)
      ]
      if len(ep_inter_ranks) > 1:
          group = dist.new_group(ep_inter_ranks)
          if rank in ep_inter_ranks:
              ep_inter_node_group = group
  
      return intra_src_rank, ep_intra_node_group, ep_inter_node_group
  
imports: import contextlib
  import os
  from typing import Any, Callable, Dict, List, Optional, Tuple
  import torch
  import torch.distributed
  import torch.nn
  import torch.nn.functional
  from colossalai.moe.manager import MOE_MANAGER
  from colossalai.tensor.moe_tensor.api import get_dp_group, get_dp_group_ranks, get_ep_size, is_moe_tensor
  from colossalai.utils import get_current_device
label_desc: Use image node_id 9 for calling a global function with example usage: create_ep_hierarchical_group(ep_group_ranks, nproc_per_node) and returns: intra_src_rank,  ep_intra_node_group,  ep_inter_node_group