还有空着的专家，为什么 MoE 仍跳过一部分 Token？

"""A deterministic top-1 router arithmetic model, not an MoE training run."""
import json
import math
from pathlib import Path


def route(rows, factor=1.0):
    if not rows or not math.isfinite(factor) or factor <= 0:
        raise ValueError("nonempty rows and positive finite factor required")
    experts = len(rows[0][2])
    if experts != 2:
        raise ValueError("this demonstration defines exactly two expert functions")
    if len({row[0] for row in rows}) != len(rows):
        raise ValueError("token IDs must be unique")
    for _, value, probabilities in rows:
        if not math.isfinite(value) or len(probabilities) != experts:
            raise ValueError("invalid input shape")
        if any(not math.isfinite(p) or p < 0 for p in probabilities):
            raise ValueError("invalid router probabilities")
        if not math.isclose(sum(probabilities), 1.0, abs_tol=1e-12):
            raise ValueError("router probabilities must sum to one")
    capacity = math.ceil(len(rows) * factor / experts)
    demand = [0] * experts
    accepted = [0] * experts
    records = []
    for token_id, value, probabilities in rows:
        # Python max keeps the first expert on a tie: our explicit policy.
        chosen = max(range(experts), key=lambda e: probabilities[e])
        demand[chosen] += 1
        keep = accepted[chosen] < capacity
        if keep:
            accepted[chosen] += 1
        expert_output = (2 * value if chosen == 0 else -value) if keep else 0.0
        output = value + probabilities[chosen] * expert_output
        records.append({"id": token_id, "expert": chosen, "kept": keep,
                        "output": round(output, 6)})
    f = [count / len(rows) for count in demand]
    mean_p = [sum(row[2][e] for row in rows) / len(rows) for e in range(experts)]
    return {"capacity": capacity, "demand": demand, "accepted": accepted,
            "dropped_ids": [r["id"] for r in records if not r["kept"]],
            "unused_slots": experts * capacity - sum(accepted),
            "balance_term_without_alpha": round(experts * sum(a*b for a,b in zip(f, mean_p)), 6),
            "records": records}


def main():
    skew = [(f"t{i}", float(i+1), [0.9, 0.1] if i < 6 else [0.1, 0.9]) for i in range(8)]
    balanced = [(f"t{i}", float(i+1), [0.9, 0.1] if i < 4 else [0.1, 0.9]) for i in range(8)]
    tied = [(f"t{i}", float(i+1), [0.5, 0.5]) for i in range(8)]
    reports = {"skew_capacity_1": route(skew), "skew_capacity_1_5": route(skew, 1.5),
               "skew_reversed": route(list(reversed(skew))),
               "balanced": route(balanced), "uniform_probabilities_tie": route(tied)}
    assert reports["skew_capacity_1"]["dropped_ids"] == ["t4", "t5"]
    assert reports["skew_capacity_1_5"]["dropped_ids"] == []
    assert reports["skew_capacity_1_5"]["unused_slots"] == 4
    assert set(reports["skew_reversed"]["dropped_ids"]) == {"t0", "t1"}
    assert reports["balanced"]["balance_term_without_alpha"] == 1.0
    assert reports["uniform_probabilities_tie"]["balance_term_without_alpha"] == 1.0
    assert len(reports["uniform_probabilities_tie"]["dropped_ids"]) == 4
    assert reports["skew_capacity_1"]["records"][4]["output"] == 5.0
    rejected = 0
    for rows, factor in [([], 1), (skew, 0), (skew, float("nan")),
                         ([("x", 1, [0.2, 0.2])], 1),
                         ([("x", 1, [1.1, -0.1])], 1),
                         ([("x", 1, [0.5, 0.5])]*2, 1)]:
        try:
            route(rows, factor)
        except ValueError:
            rejected += 1
    assert rejected == 6
    for name, result in reports.items():
        print(f"{name}: capacity={result['capacity']} demand={result['demand']} "
              f"kept={result['accepted']} dropped={result['dropped_ids']} "
              f"unused={result['unused_slots']} balance={result['balance_term_without_alpha']:.3f}")
    print("checks=14 passed; no model, training framework, GPU or API was used")
    Path("report.json").write_text(json.dumps({"scenarios": reports, "checks": 14}, indent=2), encoding="utf-8")


if __name__ == "__main__":
    main()