46. MoE Expert Parallel | MoE 专家并行

难度： Hard | 环境： CPU-first | 标签： 并行通信, MoE, Expert Parallel, All-to-All | 目标人群： 并行通信学习者

🚀 云端运行环境

本章节的实战代码可以点击以下链接在免费 GPU 算力平台上直接运行：

(国内推荐：魔搭社区免费实例)

---

本节导读

MoE 不是简单地“把参数变多”，它真正难的地方在于：token 被路由到不同专家后，专家往往分布在不同设备上，训练和推理都会引入额外通信、负载不均和 capacity 溢出。也就是说，MoE 一旦走到多设备阶段，它面对的就不再只是模型结构问题，而是一个典型的并行与通信问题。

路由器决定每个 token 进入哪些专家，负载均衡机制约束专家选择不能长期偏斜。本节继续追踪路由结果进入多设备系统后的变化：token 如何被分发、哪些 token 跨 rank、热点怎样造成容量溢出，以及专家输出为何要按原 token 顺序合并。

这一节比较不同专家放置与路由分布：同样数量的 token，均匀、本地命中和跨卡偏斜会产生不同的 overflow、通信字节和长尾风险。

关键词： expert parallel, dispatch, all-to-all

---

前置阅读

导语： 进入本节前，先能说明 Router 如何选择专家、负载均衡如何约束选择，再观察专家跨设备后如何产生 dispatch 和 all-to-all 通信。

• 06. MoE Router | MoE 路由器
• 07. MoE Load Balancing Loss | MoE 负载均衡损失

---

Step 1：从 MoE 路由到 Expert Parallel

MoE 的稀疏激活让每个 token 只计算少量专家。Expert Parallel 再把专家分给不同 rank：router 选出的专家若不在 token 的来源 rank，就必须先把 token 表示送过去；专家完成计算后，结果还要按原 token 顺序 combine。

这条链路同时引入三类变量：路由分布决定专家负载，capacity决定是否溢出，专家放置决定多少 token 需要跨卡 all-to-all。它们共同决定稀疏计算是否真正带来系统收益。

Step 2：路由、capacity 与 token dispatch

每个 token 的 routes 给出 top-k 专家候选。系统先统计 expert_loads，再依据 capacity_factor 为每个专家分配可接收路由数；超过 capacity 的部分成为 overflow。被接受的路由根据专家所属 rank 分成同 rank 与跨 rank 两类，后者才需要实际 dispatch。

因此，全部 route 的表示规模与真正跨 rank 的网络传输必须分开记录：前者是逻辑工作量，后者才取决于专家放置。图中的 dispatch 与 combine 分别是同一轮路由的去程和回程；若某个 expert 接收的 token 明显更多，会同时放大该 rank 的计算等待与通信长尾。

Step 3：比较路由分布与专家放置

比较专家并行方案时，要把三类结果分开：路由分布决定每个专家承担多少工作，容量上限决定有多少路由能够进入专家计算，专家放置决定被接受的路由中有多少需要跨 rank 传输。三者共同影响有效 token 吞吐和最慢 rank 时间。

变化	首先改变什么	需要一起观察的量
top-k 增大	总路由数与专家计算增加	overflow、dispatch bytes、有效 token 吞吐
路由偏斜	热门专家与其 rank 成为长尾	balance ratio、overflow、最长 rank 负载
capacity factor 增大	更少 token 被拒绝	预留空间、padding 与计算浪费
专家放置更本地化	跨 rank dispatch 减少	本地命中率、跨卡 bytes、负载是否反而偏斜

Step 4：CPU 机制实现——比较路由、容量与专家放置

题目区先从路由得到专家负载、容量与溢出，再将专家映射到逻辑 rank，区分本地和跨 rank token。最后依据负载、溢出和跨卡比例给出 accept / tune / reject。

任务与目的	已知条件	完成标准	测试验证
TODO 1：计算容量与溢出，判断 capacity 是否容纳当前路由	summarize_expert_parallel 已提供专家负载和平均负载	得到单专家容量与总溢出；空路由返回零	均匀、偏斜、空路由与非法配置
TODO 2：判断路由局部性，区分本地计算与跨 rank dispatch	summarize_dispatch_locality 已提供 token 来源和 expert owner	本地与跨 rank 路由之和保持不变	本地命中率与跨卡字节
TODO 3：联合容量、负载与通信风险形成决策	已提供溢出比例、负载比、跨 rank 比例和阈值	返回 accept / tune / reject	接受、调优、拒绝与高跨卡比例

from collections import Counter
from dataclasses import dataclass
import math
import torch

@dataclass
class MoEParallelSummary:
    total_tokens: int
    total_routes: int
    expert_loads: list
    capacity: int
    overflow: int
    logical_route_bytes: int
    balance_ratio: float

@dataclass
class DispatchLocalitySummary:
    total_routes: int
    local_routes: int
    cross_rank_routes: int
    cross_rank_dispatch_bytes: int
    cross_rank_roundtrip_bytes: int
    local_hit_ratio: float


def summarize_expert_parallel(routes, num_experts, hidden_size, bytes_per_elem=2, capacity_factor=1.0):
    """汇总专家负载、容量、溢出与逻辑路由字节。"""
    if num_experts <= 0 or hidden_size <= 0 or bytes_per_elem <= 0 or capacity_factor <= 0:
        raise ValueError('专家数、hidden_size、元素字节和 capacity_factor 必须为正数')

    counts = Counter()
    total_tokens = len(routes)
    total_routes = 0
    for token_routes in routes:
        for expert_id in token_routes:
            if expert_id < 0 or expert_id >= num_experts:
                raise ValueError('expert id 超出有效范围')
            counts[expert_id] += 1
            total_routes += 1

    expert_loads = [counts.get(expert_id, 0) for expert_id in range(num_experts)]
    average_load = total_routes / num_experts
    # ==========================================
    # TODO 1（容量与溢出）：由平均负载和 capacity_factor 得到容量上限。
    # capacity = ???  # 单个专家可接收的路由上限，空路由时为 0。
    # overflow = ???  # 所有专家超过 capacity 的路由数之和。
    # ==========================================
    raise NotImplementedError
    logical_route_bytes = total_routes * hidden_size * bytes_per_elem
    balance_ratio = max(expert_loads) / average_load if average_load else 0.0

    return MoEParallelSummary(
        total_tokens=total_tokens,
        total_routes=total_routes,
        expert_loads=expert_loads,
        capacity=capacity,
        overflow=overflow,
        logical_route_bytes=logical_route_bytes,
        balance_ratio=round(balance_ratio, 3),
    )


def summarize_dispatch_locality(routes, token_source_ranks, expert_owner_ranks, hidden_size, bytes_per_elem=2):
    """统计专家放置后的本地与跨 rank dispatch。

    ``token_source_ranks[i]`` 表示第 i 个 token 的来源 rank；
    ``expert_owner_ranks[e]`` 表示 expert e 所在的 rank。
    """
    if len(routes) != len(token_source_ranks):
        raise ValueError('每个 token 都需要一个来源 rank')
    if hidden_size <= 0 or bytes_per_elem <= 0:
        raise ValueError('hidden_size 与 bytes_per_elem 必须为正数')
    local_routes = 0
    cross_rank_routes = 0
    for token_idx, token_routes in enumerate(routes):
        source_rank = token_source_ranks[token_idx]
        for expert_id in token_routes:
            if expert_id not in expert_owner_ranks:
                raise ValueError('expert owner 缺少路由到的专家')
            owner_rank = expert_owner_ranks[expert_id]
            # ==========================================
            # TODO 2（路由局部性）：判断该路由是否需要跨 rank。
            # is_local = ???  # 来源 rank 与专家 owner rank 相同。
            # local_routes = ???  # 累加本地路由数。
            # cross_rank_routes = ???  # 累加跨 rank 路由数。
            # ==========================================
            raise NotImplementedError

    total_routes = local_routes + cross_rank_routes
    return DispatchLocalitySummary(
        total_routes=total_routes,
        local_routes=local_routes,
        cross_rank_routes=cross_rank_routes,
        cross_rank_dispatch_bytes=cross_rank_routes * hidden_size * bytes_per_elem,
        cross_rank_roundtrip_bytes=cross_rank_routes * hidden_size * bytes_per_elem * 2,
        local_hit_ratio=local_routes / total_routes if total_routes else 0.0,
    )


def recommend_moe_parallel_plan(
    summary: MoEParallelSummary,
    locality=None,
    max_accept_balance=1.2,
    max_accept_cross_rank=0.5,
    max_tune_overflow_ratio=0.25,
    max_tune_balance=1.8,
    max_tune_cross_rank=0.9,
) -> str:
    """根据容量、负载与跨 rank 比例返回 accept / tune / reject。"""
    if summary.total_routes == 0:
        return 'reject'
    overflow_ratio = summary.overflow / summary.total_routes
    cross_rank_ratio = (
        locality.cross_rank_routes / locality.total_routes
        if locality is not None and locality.total_routes
        else 0.0
    )
    # ==========================================
    # TODO 3（联合决策）：用显式阈值判断接受与调优区间。
    # accept_ready = ???  # 无溢出，且负载比和跨 rank 比例不超过 accept 阈值。
    # tune_ready = ???  # 三项风险均未超过 tune 阈值。
    # ==========================================
    raise NotImplementedError
    if accept_ready:
        return 'accept'
    if tune_ready:
        return 'tune'
    return 'reject'


def combine_expert_outputs(token_indices, expert_outputs, routing_weights, num_tokens):
    """按原 token 顺序，将各专家的加权输出 combine 回去。

    这是 dispatch 的反向对应步骤：同一 token 的 top-k 专家输出先乘路由权重，
    再累加到同一个 token 位置。这里提供骨架，便于把注意力放在三项并行账本 TODO 上。
    """
    if token_indices.ndim != 1 or routing_weights.ndim != 1:
        raise ValueError('token_indices 与 routing_weights 必须是一维')
    if expert_outputs.ndim != 2 or len(token_indices) != len(expert_outputs) or len(token_indices) != len(routing_weights):
        raise ValueError('专家输出、token 索引与路由权重长度必须一致')
    combined = torch.zeros(num_tokens, expert_outputs.size(-1), dtype=expert_outputs.dtype, device=expert_outputs.device)
    combined.index_add_(0, token_indices, expert_outputs * routing_weights.unsqueeze(-1))
    return combined

测试

运行下面的测试，检查你的并行负载统计、容量判断和代价估算是否正确。

def test_moe_load_capacity_contract():
    """均匀路由应得到无溢出的容量与负载账本。"""
    summary = summarize_expert_parallel([[0, 1], [1, 2], [0, 2]], 3, 4)
    assert summary.total_tokens == 3
    assert summary.total_routes == 6
    assert summary.expert_loads == [2, 2, 2]
    assert summary.capacity == 2
    assert summary.overflow == 0
    assert summary.logical_route_bytes == 48
    assert summary.balance_ratio == 1.0

def test_moe_empty_and_invalid_contract():
    """空路由返回零账本，非法专家编号或配置必须失败。"""
    empty = summarize_expert_parallel([], 2, 8)
    assert empty.total_routes == 0
    assert empty.capacity == 0
    assert empty.overflow == 0
    for args in [([[2]], 2, 8), ([[0]], 0, 8), ([[0]], 2, 0)]:
        try:
            summarize_expert_parallel(*args)
        except ValueError:
            pass
        else:
            raise AssertionError(f'非法配置应抛出 ValueError: {args}')

def test_moe_dispatch_locality_contract():
    """专家放置应区分本地命中与必须跨 rank 的 token dispatch。"""
    routes = [[0], [1], [0, 1]]
    locality = summarize_dispatch_locality(
        routes=routes,
        token_source_ranks=[0, 1, 0],
        expert_owner_ranks={0: 0, 1: 1},
        hidden_size=8,
    )
    assert locality.total_routes == 4
    assert locality.local_routes == 3
    assert locality.cross_rank_routes == 1
    assert locality.cross_rank_dispatch_bytes == 16
    assert locality.cross_rank_roundtrip_bytes == 32
    assert locality.local_hit_ratio == 0.75

def test_moe_decision_contract():
    """容量、负载和跨 rank 比例应共同改变并行决策。"""
    balanced = summarize_expert_parallel([[0], [1]], 2, 8)
    tuned = summarize_expert_parallel([[0], [0], [0], [1]], 2, 8)
    rejected = summarize_expert_parallel([[0]] * 10 + [[1]], 2, 8)
    high_cross_rank = DispatchLocalitySummary(10, 2, 8, 128, 256, 0.2)
    assert recommend_moe_parallel_plan(balanced) == 'accept'
    assert recommend_moe_parallel_plan(balanced, high_cross_rank) == 'tune'
    assert recommend_moe_parallel_plan(tuned) == 'tune'
    assert recommend_moe_parallel_plan(rejected) == 'reject'

def test_moe_combine_contract():
    """top-k 专家输出应按 token id 和路由权重回写。"""
    combined = combine_expert_outputs(
        token_indices=torch.tensor([0, 0, 1]),
        expert_outputs=torch.tensor([[1.0, 0.0], [3.0, 0.0], [0.0, 2.0]]),
        routing_weights=torch.tensor([0.25, 0.75, 1.0]),
        num_tokens=2,
    )
    assert torch.allclose(combined, torch.tensor([[2.5, 0.0], [0.0, 2.0]]))

test_moe_load_capacity_contract()
test_moe_empty_and_invalid_contract()
test_moe_dispatch_locality_contract()
test_moe_decision_contract()
test_moe_combine_contract()
print('moe expert parallel passed')

---

🛑 STOP HERE 🛑

请先尝试自己完成代码并跑通测试。
如果你正在 Colab 中运行，并且遇到困难没有思路，可以向下滚动查看参考答案。

---

参考代码与解析

代码

from collections import Counter
from dataclasses import dataclass
import math
import torch

@dataclass
class MoEParallelSummary:
    total_tokens: int
    total_routes: int
    expert_loads: list
    capacity: int
    overflow: int
    logical_route_bytes: int
    balance_ratio: float

@dataclass
class DispatchLocalitySummary:
    total_routes: int
    local_routes: int
    cross_rank_routes: int
    cross_rank_dispatch_bytes: int
    cross_rank_roundtrip_bytes: int
    local_hit_ratio: float


def summarize_expert_parallel(routes, num_experts, hidden_size, bytes_per_elem=2, capacity_factor=1.0):
    """汇总专家负载、容量、溢出与逻辑路由字节。"""
    if num_experts <= 0 or hidden_size <= 0 or bytes_per_elem <= 0 or capacity_factor <= 0:
        raise ValueError('专家数、hidden_size、元素字节和 capacity_factor 必须为正数')

    counts = Counter()
    total_tokens = len(routes)
    total_routes = 0
    for token_routes in routes:
        for expert_id in token_routes:
            if expert_id < 0 or expert_id >= num_experts:
                raise ValueError('expert id 超出有效范围')
            counts[expert_id] += 1
            total_routes += 1

    expert_loads = [counts.get(expert_id, 0) for expert_id in range(num_experts)]
    average_load = total_routes / num_experts
    # TODO 1（容量与溢出）：由平均负载和 capacity_factor 得到容量上限。
    capacity = math.ceil(average_load * capacity_factor) if total_routes else 0
    overflow = sum(max(load - capacity, 0) for load in expert_loads)
    logical_route_bytes = total_routes * hidden_size * bytes_per_elem
    balance_ratio = max(expert_loads) / average_load if average_load else 0.0

    return MoEParallelSummary(
        total_tokens=total_tokens,
        total_routes=total_routes,
        expert_loads=expert_loads,
        capacity=capacity,
        overflow=overflow,
        logical_route_bytes=logical_route_bytes,
        balance_ratio=round(balance_ratio, 3),
    )


def summarize_dispatch_locality(routes, token_source_ranks, expert_owner_ranks, hidden_size, bytes_per_elem=2):
    """统计专家放置后哪些 token 路由留在本地、哪些需要跨 rank dispatch。"""
    if len(routes) != len(token_source_ranks):
        raise ValueError('每个 token 都需要一个来源 rank')
    if hidden_size <= 0 or bytes_per_elem <= 0:
        raise ValueError('hidden_size 与 bytes_per_elem 必须为正数')
    local_routes = 0
    cross_rank_routes = 0
    for token_idx, token_routes in enumerate(routes):
        source_rank = token_source_ranks[token_idx]
        for expert_id in token_routes:
            if expert_id not in expert_owner_ranks:
                raise ValueError('expert owner 缺少路由到的专家')
            owner_rank = expert_owner_ranks[expert_id]
            # TODO 2（路由局部性）：判断该路由是否需要跨 rank。
            is_local = source_rank == owner_rank
            local_routes += int(is_local)
            cross_rank_routes += int(not is_local)

    total_routes = local_routes + cross_rank_routes
    cross_rank_dispatch_bytes = cross_rank_routes * hidden_size * bytes_per_elem
    return DispatchLocalitySummary(
        total_routes=total_routes,
        local_routes=local_routes,
        cross_rank_routes=cross_rank_routes,
        cross_rank_dispatch_bytes=cross_rank_dispatch_bytes,
        cross_rank_roundtrip_bytes=cross_rank_dispatch_bytes * 2,
        local_hit_ratio=local_routes / total_routes if total_routes else 0.0,
    )


def recommend_moe_parallel_plan(
    summary: MoEParallelSummary,
    locality=None,
    max_accept_balance=1.2,
    max_accept_cross_rank=0.5,
    max_tune_overflow_ratio=0.25,
    max_tune_balance=1.8,
    max_tune_cross_rank=0.9,
) -> str:
    """根据容量、负载与跨 rank 比例返回 accept / tune / reject。"""
    if summary.total_routes == 0:
        return 'reject'
    overflow_ratio = summary.overflow / summary.total_routes
    cross_rank_ratio = (
        locality.cross_rank_routes / locality.total_routes
        if locality is not None and locality.total_routes
        else 0.0
    )
    # TODO 3（联合决策）：用显式阈值判断接受与调优区间。
    accept_ready = (
        summary.overflow == 0
        and summary.balance_ratio <= max_accept_balance
        and cross_rank_ratio <= max_accept_cross_rank
    )
    tune_ready = (
        overflow_ratio <= max_tune_overflow_ratio
        and summary.balance_ratio <= max_tune_balance
        and cross_rank_ratio <= max_tune_cross_rank
    )
    if accept_ready:
        return 'accept'
    if tune_ready:
        return 'tune'
    return 'reject'


def combine_expert_outputs(token_indices, expert_outputs, routing_weights, num_tokens):
    """按原 token 顺序，将各专家的加权输出 combine 回去。"""
    if token_indices.ndim != 1 or routing_weights.ndim != 1:
        raise ValueError('token_indices 与 routing_weights 必须是一维')
    if expert_outputs.ndim != 2 or len(token_indices) != len(expert_outputs) or len(token_indices) != len(routing_weights):
        raise ValueError('专家输出、token 索引与路由权重长度必须一致')
    combined = torch.zeros(num_tokens, expert_outputs.size(-1), dtype=expert_outputs.dtype, device=expert_outputs.device)
    combined.index_add_(0, token_indices, expert_outputs * routing_weights.unsqueeze(-1))
    return combined

解析

TODO 1：汇总 Expert Parallel 路由账本

先校验专家编号，再由 routes 得到各专家负载。答案用平均负载和 capacity_factor 得到单专家容量，再汇总溢出与逻辑路由字节；均匀路由、空路由和非法输入测试分别验证正常、边界与失败路径。

TODO 2：区分本地与跨 rank dispatch

来源 rank 与专家 owner 相同时累计本地路由，否则累计跨 rank 路由；combine 构成回程，因此往返字节是去程的两倍。局部性测试同时检查路由守恒、本地命中率与字节账本。

TODO 3：形成容量、负载与通信联合决策

答案先计算溢出比例和跨 rank 比例，再分别应用接受阈值与调优阈值。测试覆盖无风险接受、轻度溢出调优、跨 rank 比例过高调优和严重热点拒绝；combine_expert_outputs 的独立测试验证 top-k 输出按 token id 与路由权重恢复。

Step 5（可选）：GPU Expert Parallel 机制基准

下面固定 token 数、hidden size、top-k、capacity factor 和专家放置，在真实 NCCL 环境中执行 dispatch → 本地专家计算 → combine，并比较本地优先、均匀与偏斜路由。

学习问题	CPU 已确认	GPU 要确认与后续使用
路由负载和专家放置如何改变 EP 的端到端步骤	expert load、capacity、overflow、本地命中与跨卡字节账本	比较三类路由下的跨 rank 字节、最慢 rank 时间、overflow 和峰值显存

5.1 环境与固定 workload

至少使用两张 GPU；以下配置固定专家放置和跨 rank dispatch 的 payload。

类别	固定并记录的字段	用途
运行环境	GPU 数、PyTorch / CUDA、NCCL、world size	确认 all-to-all 可启动
workload	token 数、hidden size、top-k、专家放置、dtype、warmup、repeats	固定 dispatch 规模
对照	local-heavy / balanced / skewed 路由	分开观察局部性、均衡通信和热点溢出
输出	RESULT_DIR、RUN_LABEL	分别保存三组 benchmark JSON 与 failure

# 5.1：配置真实 Expert Parallel 机制实验；默认不执行。
RUN_MODE = 'dry_run'  # 默认 dry_run；改为 real_gpu 后才启动 torchrun。
RUN_LABEL = 'default'  # 每次正式复测使用唯一标签，避免覆盖历史结果。
ENVIRONMENT_PROFILE = 'PAR-CUDA-MECHANISM'  # 至少两张 GPU，并通过该 Profile 预检。
WORKLOAD_ID = 'parallel-expert-dispatch'  # 三类路由共用的稳定 workload 身份。
WORLD_SIZE = 2  # 参与 NCCL all-to-all 的进程和 GPU 数。
NUM_EXPERTS = 4  # 固定逻辑专家数。
TOP_K = 2  # 每个 token 路由到的专家数。
CAPACITY_FACTOR = 1.25  # 单专家容量相对平均路由负载的放大系数。
TOKENS_PER_RANK = 256  # 每个 rank 输入的 token 数。
HIDDEN_SIZE = 1024  # 每条路由传输的隐藏维度。
DTYPE = 'bf16'  # 机制基准的数据类型；不支持 BF16 时改为 fp32。
RUNNER_DTYPE = 'bfloat16' if DTYPE == 'bf16' else 'float32'
ROUTING_SCENARIOS = ['local_heavy', 'balanced', 'skewed']  # 分别观察局部性、均衡和热点。
WARMUP = 5  # 计时前预热轮数。
REPEATS = 20  # 正式计时重复轮数。
RESULT_DIR = f'benchmarks/results/46_expert_parallel/{RUN_LABEL}'
print({'run_mode': RUN_MODE, 'profile': ENVIRONMENT_PROFILE, 'routing': ROUTING_SCENARIOS, 'world_size': WORLD_SIZE, 'result_dir': RESULT_DIR})

5.2 执行并保存 JSON

开启后对三种路由分别运行真实 dispatch、专家矩阵计算与 combine。除通信 payload 外，还记录 capacity、overflow、最慢 rank 时间、接收 token 分布、负载偏斜和显存。

# 5.2：执行真实 EP dispatch—compute—combine benchmark；默认关闭。
if RUN_MODE == 'real_gpu':
    import subprocess
    from pathlib import Path
    Path(RESULT_DIR).mkdir(parents=True, exist_ok=True)
    for routing in ROUTING_SCENARIOS:
        result_path = str(Path(RESULT_DIR) / f'{routing}.json')
        command = [
            'torchrun', '--standalone', '--nproc_per_node', str(WORLD_SIZE),
            'tools/run_parallel_mechanism_benchmark.py', '--mode', 'ep_dispatch',
            '--tokens-per-rank', str(TOKENS_PER_RANK), '--hidden-dim', str(HIDDEN_SIZE),
            '--num-experts', str(NUM_EXPERTS), '--top-k', str(TOP_K),
            '--capacity-factor', str(CAPACITY_FACTOR),
            '--routing', routing, '--dtype', RUNNER_DTYPE,
            '--run-label', RUN_LABEL, '--workload-id', WORKLOAD_ID,
            '--warmup', str(WARMUP), '--repeats', str(REPEATS), '--output', result_path,
        ]
        subprocess.run(command, check=True)
else:
    print('GPU Expert Parallel benchmark 默认关闭；将 RUN_MODE 改为 real_gpu 后执行。')

5.3 读取并校验结果

读取同一标签下的 local-heavy / balanced / skewed 结果，并检查公共结果字段。

# 5.3：读取三组真实 EP 机制结果。
import json
from pathlib import Path
results = {}
required = {'workload', 'hardware', 'metrics', 'evidence_level', 'failure', 'decision'}
for routing in ROUTING_SCENARIOS:
    path = Path(RESULT_DIR) / f'{routing}.json'
    if not path.exists():
        continue
    result = json.loads(path.read_text())
    missing = required - set(result)
    if missing:
        raise ValueError(f'{routing} 结果 JSON 缺少字段：{sorted(missing)}')
    results[routing] = result
print(results if results else f'尚无 Expert Parallel benchmark 结果：{RESULT_DIR}')

5.4 解释结果并形成决策

三组只改变路由分布。先排除失败运行，再比较跨 rank 字节、overflow、最慢 rank 时间、接收 token 分布和显存。

观察	结论	后续动作
failure 非空	环境或 collective 未形成有效证据	修复多进程 / NCCL 后重跑
local-heavy 的跨 rank 字节下降	专家放置改善通信局部性	检查是否引入新的专家负载偏斜
skewed 的 overflow、最慢 rank 时间或负载离散度升高	路由热点进入关键路径	调整 capacity、router 或专家放置
三组字段完整且重复结果稳定	获得 EP 机制级证据	在项目页复测真实模型吞吐与长尾

相关阅读

完成 expert load、capacity、overflow 和 all-to-all 账本后，可以继续阅读 MoE 论文、分布式实现与真实 benchmark。

• Switch Transformers 原论文
• Megatron-LM 官方仓库
• 79. Distributed Parallel Benchmark | 分布式并行 Benchmark
• 80. MoE Expert Parallel Benchmark | MoE 专家并行 Benchmark
• 81. Distributed Inference Project | 分布式推理项目