01 / 本课学习路线
本课学习路线
阅读与推演约 100 分钟，练习约 55 分钟，进阶练习另需约 20 分钟
02 / 学习目标与先修自测
学完本课你能做什么，以及开始前需要会什么
三道题都是「规则题」：题目把规则逐条写清，程序按条执行。难点不在算法，而在把并列规则写进排序键、把容量判断放在写入之前、把合并公式的两边都缩放。
| 学完后能做的事 | 正文位置 | 检查方式 |
|---|---|---|
| 写出流水线总时延公式并解释两项各自的含义 | 第 04 节 | 自查第 2 条、练习 1 |
| 手工推演 AI040 样例 2 的同分处理与丢弃结果 | 第 05、06 节 | 自查第 3 条、练习 2 |
| 说出词元选专家与专家选词元的方向差别 | 第 05 节 | 自查第 4 条 |
| 写出 (m, l) 合并公式并说明为什么两边都要缩放 | 第 05 节补充学习、第 06 节 | 自查第 5 条、练习 3、5 |
| 写出三类题（瓶颈、路由、NMS）中并列处理的排序键 | 第 03、05 节 | 自查第 6 条、练习 4 |
先修自测：下面 5 题请先自己写答案，再展开对照。答不出的按括号里的位置补看再回来。本课假定你已完成模块 2 的「复合排序与并列规则」、模块 6 的「浮点精度与稳定 softmax」和模块 8 的「卷积尺寸、交并比（IoU）与非极大值抑制（NMS）」。
| 题号 | 题目 | 补看位置 |
|---|---|---|
| 自测 1 | t = [6, 2, 6, 6]：max(t) 是多少？t.index(max(t)) 是多少？要输出 1 基编号该怎么改？ | 列表常用操作 |
| 自测 2 | g = [1, 2, 9]，sorted(range(3), key=lambda x: (-g[x], x))[:2] 得到什么？ | 复合排序与并列规则 |
| 自测 3 | (10**9 - 1) * 10**6 + 2 * 10**6 在 Python 里会溢出吗？换成 C++ 的 int 呢？ | 数字与运算符 |
| 自测 4 | math.exp(1000) 会发生什么？math.exp(-1000) 呢？ | 浮点精度与稳定 softmax |
| 自测 5 | math.log(0) 会发生什么？合并状态 (m, l) 的 l 什么时候可能是 0？ | 异常与错误处理 |
展开先修自测答案
自测 1：max(t) = 6，t.index(6) = 0——index 返回第一个匹配的下标，正好是「并列取编号最小」；输出 1 基编号加 1，得 1。这就是 AI041 题库用例「多瓶颈并列」的答案 44 1。
自测 2：按 −g 升序即 g 降序：9（下标 2）、2（下标 1）、1（下标 0）→ [2, 1, 0]，切片取前 2 得 [2, 1]。这就是 AI040 示例 1 里第 2 个词元的候选顺序。
自测 3：Python 的整数是任意精度，不会溢出，结果 1000000001000000（约 10¹⁵）；C++ 的 32 位 int 最大约 2.1×10⁹，会溢出，要用 long long（Java 用 long）。AI041 题库里 m=10⁹、t=10⁶ 的那组用例，答案就是这个数。
自测 4：math.exp(1000) 抛出 OverflowError（double 最大约 1.8×10³⁰⁸，e^{709.8} 就到顶了）；math.exp(-1000) 不报错，返回 0.0（下溢）。AI038 示例 2 的 ±10⁹ 就是要让朴素写法在这里出错。
自测 5：math.log(0) 抛出 ValueError。l 是 Σe^{s_i − m}，只要 m 取的是区间里真实存在的最大值，至少有一项 e^0 = 1，l ≥ 1，不会是 0；若把初始状态写成 m=0、l=0，再并入一个很负的数，e^{s − 0} 下溢成 0，就会走到 log(0)（第 08 节错误表最后一行）。
03 / 概念与术语
阶段、微批次、瓶颈；专家、候选、容量、丢弃；可合并状态
三道题三组词，公共的只有一条：并列规则不靠默认顺序，写进排序键或比较符号。
| 术语 | 含义 | 代码里的位置 |
|---|---|---|
阶段（stage）与耗时 t[i] | 流水线的第 i 段，编号 1..p；同一时刻只处理一个微批次 | stages 列表，下标 0 起、输出时 +1 |
微批次（micro-batch）与 m | 一小批数据；m 个按编号依次进入，阶段之间缓冲区不限 | 公式里的 m |
| 瓶颈阶段 | 耗时最大的阶段，并列取编号最小；总时延的第二项由它决定 | if stages[i] > stages[best] |
| 总完成时间 | Σt_i + (m−1)·max(t_i)：首个微批次走完全程，之后每隔 max(t) 出一个 | pipeline_time |
专家（expert）与得分 g[t][x] | e 个专家，编号 0..e−1；词元 t 对专家 x 的得分 | g[t][x] |
| 候选（前 k） | 按（得分降序，专家编号升序）排序后的前 k 个；只在这 k 个里尝试 | sorted(...)[:k] |
| 容量 c 与负载 | 每个专家最多接收 c 个词元；负载 = 已接收数，写入前判断 < c | load[x] < c |
丢弃（dropped） | 候选全部被跳过的词元；输出 t: -，并进最后一行 | if not got: dropped.append(t) |
| 词元选专家 / 专家选词元 | 外层循环是词元（AI040）还是专家；方向不同，循环主体完全不同 | 外层 for t in range(n) |
| 可合并状态 (m, l) | m = 区间最大值，l = Σe^{s_i − m}；合并时取新 m，两边各乘 e^{旧 m − 新 m} | merge |
| 对数求和指数（log-sum-exp，LSE） | ln Σe^{s_i} = m + ln l；就是 softmax 的归一化常数取对数 | m + math.log(total) |
三道题的编号基准不同：AI041 的阶段编号从 1 开始（输入的 s 也是 1 基），AI040 的词元与专家编号从 0 开始，AI038 的区间 [l, r] 是 1 基闭区间。读入后先在纸上写下「代码下标 = 题目编号 − 几」，输出时再换回去。
04 / 流水线时延
从三个微批次的时间线推出总时延公式
AI041 示例 1：p=3、m=3、t = [2, 5, 1]。先按题目规则把每个微批次在每个阶段的开始与结束时刻排出来，再从表里读出公式。
| 微批次 | 阶段 1（t=2） | 阶段 2（t=5） | 阶段 3（t=1） |
|---|---|---|---|
| 1 | 0–2 | 2–7 | 7–8 |
| 2 | 2–4 | 7–12 | 12–13 |
| 3 | 4–6 | 12–17 | 17–18 |
规则只有两条：阶段 i 空闲且微批次 j 已完成阶段 i−1 时立刻开始；同一阶段同一时刻只处理一个微批次。微批次 2 在时刻 4 就做完了阶段 1，却要等到 7 才能进阶段 2——因为阶段 2 正在处理微批次 1。之后每个微批次进阶段 2 的时刻相隔 5 = max(t)，离开阶段 3 的时刻也相隔 5。总完成时间 18 = 8（微批次 1 走完全程 2+5+1）+ 2×5（后面两个微批次各隔 5）。
一般式：总完成时间 = Σt_i + (m−1)·max(t_i)。第一项是第一个微批次从头到尾的时间；第二项来自「最慢的阶段每隔 max(t) 才能放行一个微批次」，最后一个微批次比第一个晚 (m−1)·max(t) 离开。这个式子要求阶段之间缓冲区不限（题目已保证）；p=1 时退化为 m·t，示例 2 的 1 10 / 7 / 1 3 得 70，优化后 4×10 = 40。
| 微批次 | 阶段 1（t=2） | 阶段 2（t=3） | 阶段 3（t=1） |
|---|---|---|---|
| 1 | 0–2 | 2–5 | 5–6 |
| 2 | 2–4 | 5–8 | 8–9 |
| 3 | 4–6 | 8–11 | 11–12 |
总完成时间 12 = 6 + 2×3。瓶颈要重新扫描：这里仍是阶段 2（3 > 2 > 1）；题库用例 3 100 / 10 9 1 / 1 5 优化后 t = [5, 9, 1]，瓶颈从 1 号转移到 2 号，输出 1010 1 / 906 2——沿用旧编号会输出 906 1（第 08 节）。并列时取编号最小：4 5 / 6 2 6 6 / 4 1 优化前三个 6 取 1 号，优化后 t = [6, 2, 6, 5] 仍取 1 号，输出 44 1 / 43 1。
为什么不能逐批模拟
m 到 10⁹、t 到 10⁶：(m−1)·max(t) 约 10¹⁵。逐批逐阶段模拟需要 O(m·p) 次更新，工作量随微批次数 m 增长；改用公式，只需扫描 p 个阶段，时间复杂度为 O(p)。Python 的整数是任意精度不会溢出；C++ 用 long long、Java 用 long（64 位）。
05 / 调度与路由
逐条实现三类规则
本课的要求是将每条题目规则落实为明确的实现步骤。每条规则配一个针对性的验证输入，写完先运行验证再提交。
| 规则 | 代码 | 验证输入（题库用例） | 期望 |
|---|---|---|---|
| 按词元编号顺序处理，先处理的先占容量 | 外层 for t in range(n)；load 跨词元累计 | 3 2 1 1 / 9 8 / 9 8 / 9 8 | 0: 0 / 1: - / 2: - / dropped: 1 2 |
| 候选 = 按（得分降序，编号升序）取前 k，只在前 k 里尝试，满了跳过不顺延 | sorted(range(e), key=lambda x: (-g[t][x], x))[:k] | 4 3 2 1 / 9 8 7 ×4 | 0: 0 1 / 1: - / 2: - / 3: - / dropped: 1 2 3（词元 1 不会顺延到专家 2） |
候选全部跳过 → 丢弃；输出 t: -，最后一行列出全部被丢弃的编号 | if not got: dropped.append(t) | 3 2 2 0 / …（c=0） | 三行 -，dropped: 0 1 2 |
MoE 同分样例（AI040 样例 2）
AI040 样例 2：2 个词元、3 个专家、k=1、c=1，分数全是 5 t0：候选排序 = 按(得分降序, 编号升序) → [E0]，E0 未满 → t0 进 E0 t1：候选还是 [E0]（k=1 只看第一名），E0 已满 → 不向第 2 名顺延 → t1 整个丢弃 输出：0: 0 ／ 1: - ／ dropped: 1
专家选词元是反方向：外层循环换成专家，每个专家从全部词元里按得分挑最多 c 个；一个词元可能被多个专家挑中，也可能没有任何专家要它。两种方向的输出结构都不一样，读题第一件事是确认方向——AI040 的规则 1「按词元编号从小到大依次处理」已经说明是词元选专家。
从空文件实现流水线总时延函数（pipeline_time）与专家路由函数（moe_route）
Python# 练习模板：标有「待完成」的部分需要补全
import sys
def pipeline_time(stages, m):
# 待完成：sum(t) + (m-1) * max(t)；瓶颈并列取编号最小
# m 到 1e9、t 到 1e6：乘积约 1e15；Python 的整数是任意精度不会溢出，C++ 用 long long、Java 用 long（64 位）
...
def moe_route(g, k, c):
# 待完成：词元按编号顺序处理；候选=按（得分降序，专家编号升序）取前 k；
# 只在前 k 里尝试，满了跳过不顺延；全部跳过 → 丢弃
...
def main():
data = sys.stdin.read().split()
# 待完成：AI041 输出两行「总时间 瓶颈编号」；AI040 注意 t: - 与 dropped: - 的格式
...
main()AI041 的数量级判断：m 到 10⁹、t 到 10⁶，(m−1)·max(t) 约 10¹⁵——m 最大为 10⁹，逐批模拟无法满足数据范围，应使用总时延公式直接计算；C++ 用 long long、Java 用 long（64 位）。AI040 的容量判断要放在写入之前：写进去才检查，超额词元已经污染状态。
先把并列规则写进排序键
AI041 瓶颈并列取编号最小；AI040 同分取专家编号小。「只按得分排序」在同分用例上顺序不定——排序键必须显式写成 (−得分, 专家编号)，这和模块 8 · 第 1 课 里 NMS 的 (−得分, 下标) 采用相同的并列处理方式：主键之外再加一个编号键，使顺序确定。
补充学习（选学）FlashAttention 分块归约与普通 softmax 的区别约 6 分钟在线 (m, l) 状态的合并公式与结合律
普通实现先算全行分数再一次 softmax；分块归约维护在线状态 (m, l)：新块并入时 m_new = max(m_old, m_blk)，l_new = l_old·e^{m_old−m_new} + l_blk·e^{m_blk−m_new}。两边都要按新 m 重新缩放——只缩一边就错。
AI038 的样例是 ±10⁹ 的分数：朴素 Σe^s 在正数超过约 709 时上溢为 inf、很小的指数项可能下溢为 0，造成严重精度损失，而 (m, l) 状态全程有界。LSE(1,2,3) = 3 + ln(1+e^{−1}+e^{−2}) = 3.4076，先用小例子验证合并函数，再上大数值。需要掌握的是「按给定合并公式一步不差地执行」——对 LSE 这个量，它与先取全局最大再一次求和在实数意义下完全一致，浮点只有末位差异，题目已保证四位小数结果相同。要得到完整的注意力输出，还要再维护一个按同样比例重缩放的加权输出累加器——AI038 只要求 LSE 这一层。
06 / 题面示例逐步
AI041 两个示例、AI040 示例 1、AI038 两个示例
AI041 的时间线表在第 04 节。AI040「混合专家层的容量路由」：第一行 n e k c，随后 n 行、每行 e 个整数得分；输出 n 行 t: 专家列表 与最后一行 dropped: 编号列表，没有时用 -。示例 1：3 3 2 2 / 9 5 1 / 8 7 2 / 1 2 9 → 0: 0 1 / 1: 0 1 / 2: 2 / dropped: -。
| 词元 | 得分 | 排序后（得分降序、编号升序） | 前 k 候选 | 处理前负载 | 逐个尝试 | 处理后负载 | 输出行 |
|---|---|---|---|---|---|---|---|
| 0 | 9 5 1 | 0, 1, 2 | 0, 1 | [0, 0, 0] | 0：0 < 2 分配；1：0 < 2 分配 | [1, 1, 0] | 0: 0 1 |
| 1 | 8 7 2 | 0, 1, 2 | 0, 1 | [1, 1, 0] | 0：1 < 2 分配；1：1 < 2 分配 | [2, 2, 0] | 1: 0 1 |
| 2 | 1 2 9 | 2, 1, 0 | 2, 1 | [2, 2, 0] | 2：0 < 2 分配；1：2 不小于 2 跳过 | [2, 2, 1] | 2: 2 |
没有词元被丢弃，最后一行 dropped: -。第 3 个词元的第 2 名专家 1 已满，被跳过——但不会顺延到排在第 3 的专家 0（它也满了，而且不在前 k 里）。示例 2 的推演见第 05 节。
AI038「注意力分数的稳定区间求和」：第一行 n q，第二行 n 个实数，随后 q 行 l r（1 基闭区间）；每个查询输出 ln(e^{s_l} + … + e^{s_r})，保留 4 位小数。示例 1：3 2 / 1.0000 2.0000 3.0000 / 1 3 / 2 2 → 3.4076 / 2.0000；示例 2：2 1 / 1000000000.0000 -1000000000.0000 / 1 2 → 1000000000.0000。
| 并入 | 合并前 (m, l) | 新 m | l 的两项 | 合并后 (m, l) |
|---|---|---|---|---|
| s_1 = 1 | 空 | 1 | — | (1, 1) |
| s_2 = 2 | (1, 1) | 2 | 1·e^{1−2} + 1·e^{2−2} = 0.367879 + 1 | (2, 1.367879) |
| s_3 = 3 | (2, 1.367879) | 3 | 1.367879·e^{2−3} + 1·e^{3−3} = 0.503215 + 1 | (3, 1.503215) |
答案 = m + ln l = 3 + ln 1.503215 = 3 + 0.407606 = 3.407606 → 3.4076。查询 [2, 2] 只有一个元素，(2, 1) → 2 + ln 1 = 2.0000。示例 2：并入 10⁹ 得 (10⁹, 1)；再并入 −10⁹，新 m 仍是 10⁹，l = 1·e^0 + 1·e^{−2×10⁹} = 1 + 0 = 1，答案 10⁹ + 0 = 1000000000.0000——e^{−2×10⁹} 下溢成 0 是无害的，它本来就小到对 4 位小数没有影响；朴素写法先算 e^{10⁹} 就直接上溢（第 08 节）。
题库里另有两组可以手算核对的边界：8 个相同的 2.5，查询 [1, 8] 得 2.5 + ln 8 = 4.5794、查询 [3, 5] 得 2.5 + ln 3 = 3.5986；单元素 −123.4567 的答案就是它本身。
07 / 从三个函数到程序
参考实现与三份完整程序，每一步落在哪几行
先用断言分别验证 pipeline_time、moe_route 与 merge，再各自加上读入和输出组成完整程序。
| 步骤 | AI041 | AI040 | AI038 |
|---|---|---|---|
| 读入 | p m；p 个 t；s x | n e k c；n 行 e 个得分；k > e 先输出 invalid | n q；n 个实数；q 行 l r |
| 核心 | pipeline_time：首个最大值下标 + 公式 | moe_route：排序键 → 切前 k → 写入前判容量 | merge：取新 m，两边缩放；lse_range 从左到右合并 |
| 并列 / 边界 | > 才换，保留最小编号 | 键 (−得分, 编号)；load[x] < c | 空状态 l = 0；单元素 l = 1 |
| 输出 | 两行 总时间 编号（1 基） | n 行 t: … 或 t: -；dropped: … 或 dropped: - | q 行 4 位小数 |
展开参考实现 1：pipeline_time 与 moe_route（自带断言；先自己写完再对照）
pipeline_time 与 moe_route 的参考实现（自带断言）
Pythondef pipeline_time(stages, m):
# 总完成时间 = Σt + (m−1)·max(t)；瓶颈并列取编号最小（编号从 1 起）
best = 0
for i in range(1, len(stages)):
if stages[i] > stages[best]:
best = i
return sum(stages) + (m - 1) * stages[best], best + 1
def moe_route(g, k, c):
# 词元按编号顺序处理；候选 = 按（得分降序，专家编号升序）取前 k；满了跳过不顺延；全部跳过 → 丢弃
n, e = len(g), len(g[0])
load = [0] * e
assigned, dropped = [], []
for t in range(n):
cand = sorted(range(e), key=lambda x: (-g[t][x], x))[:k]
got = []
for x in cand:
if load[x] < c: # 写入之前先判断容量
load[x] += 1
got.append(x)
assigned.append(got)
if not got:
dropped.append(t)
return assigned, dropped
# AI041 题面示例：优化前 / 后；p=1 退化为串行；并列取编号最小；优化后瓶颈转移；练习 1 的 m=4
assert pipeline_time([2, 5, 1], 3) == (18, 2)
assert pipeline_time([2, 3, 1], 3) == (12, 2)
assert pipeline_time([7], 10) == (70, 1)
assert pipeline_time([6, 2, 6, 6], 5) == (44, 1)
assert pipeline_time([10, 9, 1], 100) == (1010, 1) and pipeline_time([5, 9, 1], 100) == (906, 2)
assert pipeline_time([2, 5, 1], 4) == (23, 2)
# AI040 题面示例 1 / 2；只在前 k 里尝试（4 个词元都想要专家 0、1，容量 1）；c=0 全部丢弃；练习 2 的 c=1
assert moe_route([[9, 5, 1], [8, 7, 2], [1, 2, 9]], 2, 2) == ([[0, 1], [0, 1], [2]], [])
assert moe_route([[5, 5, 5], [5, 5, 5]], 1, 1) == ([[0], []], [1])
assert moe_route([[9, 8, 7]] * 4, 2, 1) == ([[0, 1], [], [], []], [1, 2, 3])
assert moe_route([[1, 2], [3, 4], [5, 6]], 2, 0) == ([[], [], []], [0, 1, 2])
assert moe_route([[9, 5, 1], [8, 7, 2], [1, 2, 9]], 2, 1) == ([[0, 1], [], [2]], [1])断言覆盖 AI041 两个示例、并列取小、瓶颈转移、练习 1 的 m=4；AI040 两个示例、只在前 k 里尝试、c=0、练习 2 的 c=1。
展开参考实现 2：merge 与 lse_range（自带断言；进阶练习）
merge 与 lse_range 的参考实现（自带断言）
Pythonimport math
def merge(a, b):
# 状态 (m, l)：m 是区间最大值，l = Σ e^{s_i − m}；l = 0 表示空状态
m1, l1 = a
m2, l2 = b
if l1 == 0.0:
return b
if l2 == 0.0:
return a
m = max(m1, m2)
return m, l1 * math.exp(m1 - m) + l2 * math.exp(m2 - m) # 两边都按新 m 缩放
def lse_range(s, l, r):
# 区间 [l, r]（1 基、闭区间）从左到右逐个合并；答案 = m + ln l
state = (0.0, 0.0)
for i in range(l - 1, r):
state = merge(state, (s[i], 1.0))
m, total = state
return m + math.log(total)
R = lambda v: round(v, 4)
# AI038 题面示例 1 / 2
assert R(lse_range([1.0, 2.0, 3.0], 1, 3)) == 3.4076 and R(lse_range([1.0, 2.0, 3.0], 2, 2)) == 2.0
assert R(lse_range([1e9, -1e9], 1, 2)) == 1e9
# 单元素就是它本身；k 个相同值 = s + ln k
assert R(lse_range([-123.4567], 1, 1)) == -123.4567
assert R(lse_range([2.5] * 8, 1, 8)) == R(2.5 + math.log(8))
# 练习 3：三个数都乘 1000，朴素 Σexp 会上溢，稳定状态给 3000.0000
assert R(lse_range([1000.0, 2000.0, 3000.0], 1, 3)) == 3000.0
# 结合律：先合前两个再并第三个 == 先合后两个再并第一个（分治也能用同一个 merge）
a, b, c = (1.0, 1.0), (2.0, 1.0), (3.0, 1.0)
left, right = merge(merge(a, b), c), merge(a, merge(b, c))
assert left[0] == right[0] and abs(left[1] - right[1]) < 1e-12
# 合并的中间状态：(1, 1) → (2, 1.367879) → (3, 1.503215)
st = merge((1.0, 1.0), (2.0, 1.0))
assert st[0] == 2.0 and abs(st[1] - 1.367879) < 1e-6
st = merge(st, (3.0, 1.0))
assert st[0] == 3.0 and abs(st[1] - 1.503215) < 1e-6断言覆盖 AI038 两个示例、单元素、相同值 = s + ln k、练习 3 的 ×1000、结合律，以及示例 1 三次合并的中间状态。
展开完整参考程序 1：AI041 推理流水线总耗时与瓶颈
完整程序：AI041（标准输入 → 标准输出）
Pythonimport sys
def pipeline_time(stages, m):
# 总完成时间 = Σt + (m−1)·max(t)；瓶颈 = 耗时最大的阶段，并列取编号最小（编号从 1 起）
best = 0
for i in range(1, len(stages)):
if stages[i] > stages[best]: # 严格大于才换，自然保留最小编号
best = i
total = sum(stages) + (m - 1) * stages[best]
return total, best + 1
def main():
data = sys.stdin.read().split()
p, m = int(data[0]), int(data[1])
t = [int(v) for v in data[2:2 + p]]
s, x = int(data[2 + p]), int(data[3 + p])
total, bott = pipeline_time(t, m)
print(total, bott)
t[s - 1] -= x # 优化：阶段 s 的耗时减少 x
total, bott = pipeline_time(t, m) # 重新计算，不能沿用旧瓶颈
print(total, bott)
main()适用于 p 到 10⁵、m 到 10⁹ 的全部数据范围。优化后重新调用 pipeline_time，不沿用旧瓶颈编号。
展开完整参考程序 2：AI040 混合专家层的容量路由
完整程序：AI040（标准输入 → 标准输出）
Pythonimport sys
def moe_route(g, k, c):
# 词元按编号顺序处理；候选 = 按（得分降序，专家编号升序）取前 k；
# 只在前 k 个里尝试，满了跳过不顺延；全部跳过 → 丢弃
n, e = len(g), len(g[0])
load = [0] * e # 每个专家已接收的词元数
assigned, dropped = [], []
for t in range(n):
cand = sorted(range(e), key=lambda x: (-g[t][x], x))[:k]
got = []
for x in cand:
if load[x] < c: # 写入之前先判断容量
load[x] += 1
got.append(x)
assigned.append(got)
if not got:
dropped.append(t)
return assigned, dropped
def main():
data = sys.stdin.read().split()
n, e, k, c = (int(v) for v in data[:4])
if k > e:
print("invalid")
return
vals = [int(v) for v in data[4:4 + n * e]]
g = [vals[t * e:(t + 1) * e] for t in range(n)]
assigned, dropped = moe_route(g, k, c)
lines = []
for t in range(n):
lines.append(f"{t}: " + (" ".join(str(x) for x in assigned[t]) if assigned[t] else "-"))
lines.append("dropped: " + (" ".join(str(t) for t in dropped) if dropped else "-"))
print("\n".join(lines))
main()适用于 n、e 到 200 的全部数据范围。k > e 在读得分矩阵之前就输出 invalid 并返回；- 只在列表为空时输出。
展开完整参考程序 3：AI038 注意力分数的稳定区间求和（进阶练习）
完整程序：AI038（标准输入 → 标准输出）
Pythonimport math
import sys
def merge(a, b):
# 状态 (m, l)：m 是区间最大值，l = Σ e^{s_i − m}；合并时两边都按新的 m 重新缩放
m1, l1 = a
m2, l2 = b
if l1 == 0.0: # 空状态：直接取另一边
return b
if l2 == 0.0:
return a
m = max(m1, m2)
return m, l1 * math.exp(m1 - m) + l2 * math.exp(m2 - m)
def lse_range(s, l, r):
# 从左到右逐个合并区间 [l, r]（1 基、闭区间）；答案 = m + ln l
state = (0.0, 0.0) # 空状态用 l = 0 标记
for i in range(l - 1, r):
state = merge(state, (s[i], 1.0))
m, total = state
return m + math.log(total)
def main():
data = sys.stdin.read().split()
n, q = int(data[0]), int(data[1])
s = [float(v) for v in data[2:2 + n]]
out = []
pos = 2 + n
for _ in range(q):
l, r = int(data[pos]), int(data[pos + 1])
pos += 2
out.append(f"{lse_range(s, l, r):.4f}")
print("\n".join(out))
main()适用于 n 到 20000、q 到 50 的全部数据范围，每个查询线性合并即可。空状态用 l = 0 标记，首次合并直接取另一边，不需要 −inf。
08 / 边界、反例与复杂度
用小输入检查规则，用工作量检查复杂度
对照表中的输入、错误表现与正确输出定位差异；输入中的斜杠表示换行。逐批模拟一行比较的是工作量，使用小规模变式核对公式即可，不需要运行十亿规模模拟。
| 错误做法 | 输入 | 错误输出 | 正确输出 | 判题结果 |
|---|---|---|---|---|
| AI041 总时间写成 m·Σt（不重叠） | 示例 1 3 3 / 2 5 1 / 2 2 | 24 2 / 18 2 | 18 2 / 12 2 | 答案错误（WA） |
| AI041 优化后沿用旧瓶颈编号 | 3 100 / 10 9 1 / 1 5 | 1010 1 / 906 1 | 1010 1 / 906 2 | 答案错误（WA） |
AI041 并列写成 >=（取大编号） | 4 5 / 6 2 6 6 / 4 1 | 44 4 / 43 3 | 44 1 / 43 1 | 答案错误（WA） |
| AI041 输出 0 基编号 | 示例 1 | 18 1 / 12 1 | 18 2 / 12 2 | 答案错误（WA） |
| AI041 逐个微批次模拟 | 2 1000000000 / 1000000 1000000 / 1 1 | 每次计算总时间需要 2×10⁹ 次阶段更新 | 1000000001000000 1 / 1000000000999999 2 | 应改用 O(p) 公式 |
| AI040 满了就顺延到第 k 名之后 | 4 3 2 1 / 9 8 7 ×4 | 0: 0 1 / 1: 2 / 2: - / 3: - / dropped: 2 3 | 0: 0 1 / 1: - / 2: - / 3: - / dropped: 1 2 3 | 答案错误（WA） |
AI040 同分取大编号 (-g, -x) | 示例 2 2 3 1 1 / 5 5 5 / 5 5 5 | 0: 2 / 1: - / dropped: 1 | 0: 0 / 1: - / dropped: 1 | 答案错误（WA） |
AI040 容量判断写成 <= c | 3 2 1 1 / 9 8 / 9 8 / 9 8 | 0: 0 / 1: 0 / 2: - / dropped: 2 | 0: 0 / 1: - / 2: - / dropped: 1 2 | 答案错误（WA） |
| AI040 不判 k > e | 2 2 3 1 / 1 2 / 3 4 | 0: 1 0 / 1: - / dropped: 1 | invalid | 答案错误（WA） |
AI040 没丢弃时 dropped: 后不打 - | 示例 1 | 最后一行 dropped: | 最后一行 dropped: - | 答案错误（WA） |
| AI038 朴素 ln Σe^{s_i} | 示例 2 2 1 / 1000000000.0000 -1000000000.0000 / 1 2 | OverflowError，无输出 | 1000000000.0000 | 运行错误（RE） |
AI038 合并时只缩放新块 l1 + l2·e^{m2−m} | 示例 1 | 4.0986 / 2.0000 | 3.4076 / 2.0000 | 答案错误（WA） |
| AI038 答案忘加 m（只输出 ln l） | 示例 1 | 0.4076 / 0.0000 | 3.4076 / 2.0000 | 答案错误（WA） |
AI038 区间当成 0 基 range(l, r + 1) | 示例 1 | IndexError，无输出 | 3.4076 / 2.0000 | 运行错误（RE） |
| AI038 初始状态写成 (0, 0) 且不判空 | 3 1 / -800000000.0000 -800000001.0000 -799999999.5000 / 1 3 | ValueError: math domain error（e^{−8×10⁸} 下溢成 0 后取 log） | -799999998.8959 | 运行错误（RE） |
第 12 行：只缩放新块时 l 变成 1 + 1 = 2、再 2 + 1 = 3，3 + ln 3 = 4.0986——旧块的 e^{m_old − m_new} 被漏掉。第 5 行的逐批模拟在小输入上可以得到正确结果，但其更新次数随 m 增长；小样例正确不代表复杂度适合本题。用小规模输入比较模拟与公式即可，不必运行十亿微批次模拟。其他行可按给定输入或题面示例核对修改前后的输出。
| 做法 | 时间 | 本课规模下 |
|---|---|---|
| AI041 公式 | O(p) | p ≤ 10⁵，公式扫描阶段；逐批模拟 O(m·p) 的更新次数随 m 增长 |
| AI040 每个词元排序 | O(n·e·log e) | 按每个词元对 e 个分数排序，再遍历前 k 个；n、e ≤ 200 |
| AI038 每查询线性合并 | O(q·n) | 最多 50×19999 次非空合并，每次两次 exp，合计 1,999,900 次 exp；按给定范围估算运算量 |
09 / 渐进练习与参考答案
跟做 → 改一个条件 → 独立实现 → 迁移
每题先在纸上或文件里做完，再展开答案。
练习 1（跟做）：把 AI041 示例 1 的 m 改成 4。先按第 04 节的格式补出第 4 个微批次那一行时间线，再用公式算优化前与优化后的总完成时间。
展开练习 1 答案
第 4 行：阶段 1 6–8、阶段 2 17–22（要等微批次 3 在 17 离开阶段 2）、阶段 3 22–23。公式：8 + 3×5 = 23；优化后 6 + 3×3 = 15。输出 23 2 / 15 2（参考实现里有 m=4 的断言）。每多一个微批次，总时间多一个 max(t)，这就是公式第二项的含义。
练习 2（改一个条件）：把 AI040 示例 1 的容量 c 从 2 改成 1，输出四行各是什么？哪个词元被丢弃，为什么它不能去专家 2？
展开练习 2 答案
词元 0 的候选 0、1 都空，占满：0: 0 1，负载 [1, 1, 0]。词元 1 的候选也是 0、1，都满、都跳过，没有分配：1: -。词元 2 的候选是 2、1：2 空闲分配，1 满跳过：2: 2。最后一行 dropped: 1。词元 1 不能去专家 2，因为它对专家 2 的得分 2 排在第 3，不在前 k=2 的候选里——规则 2 明确「不向第 k 名之后的专家顺延」。
练习 3（改一个条件）：把 AI038 示例 1 的三个数都乘 1000（变成 1000、2000、3000），查询 [1, 3] 的答案是多少？朴素写法 ln(e^{1000} + e^{2000} + e^{3000}) 会怎样？
展开练习 3 答案
合并到最后 m = 3000，l = e^{−2000} + e^{−1000} + 1 = 1（前两项下溢成 0，本来也远小于 10⁻⁴），答案 3000 + ln 1 = 3000.0000。朴素写法第一步 math.exp(1000) 就抛出 OverflowError——double 在 e^{709.8} 附近到顶。稳定状态里指数永远是 s_i − m ≤ 0，只会下溢不会上溢，而下溢掉的项对 4 位小数没有影响。
练习 4（独立实现）：不看第 07 节，完成模板里的 pipeline_time 与 moe_route，并让它们通过：AI041 两个示例、[6, 2, 6, 6] 并列取 1、[10, 9, 1] 减 5 后瓶颈转到 2；AI040 两个示例、四个词元都想要专家 0 和 1 而容量为 1。
展开练习 4 答案
pipeline_time 与 moe_route 的参考实现（自带断言）
Pythondef pipeline_time(stages, m):
# 总完成时间 = Σt + (m−1)·max(t)；瓶颈并列取编号最小（编号从 1 起）
best = 0
for i in range(1, len(stages)):
if stages[i] > stages[best]:
best = i
return sum(stages) + (m - 1) * stages[best], best + 1
def moe_route(g, k, c):
# 词元按编号顺序处理；候选 = 按（得分降序，专家编号升序）取前 k；满了跳过不顺延；全部跳过 → 丢弃
n, e = len(g), len(g[0])
load = [0] * e
assigned, dropped = [], []
for t in range(n):
cand = sorted(range(e), key=lambda x: (-g[t][x], x))[:k]
got = []
for x in cand:
if load[x] < c: # 写入之前先判断容量
load[x] += 1
got.append(x)
assigned.append(got)
if not got:
dropped.append(t)
return assigned, dropped
# AI041 题面示例：优化前 / 后；p=1 退化为串行；并列取编号最小；优化后瓶颈转移；练习 1 的 m=4
assert pipeline_time([2, 5, 1], 3) == (18, 2)
assert pipeline_time([2, 3, 1], 3) == (12, 2)
assert pipeline_time([7], 10) == (70, 1)
assert pipeline_time([6, 2, 6, 6], 5) == (44, 1)
assert pipeline_time([10, 9, 1], 100) == (1010, 1) and pipeline_time([5, 9, 1], 100) == (906, 2)
assert pipeline_time([2, 5, 1], 4) == (23, 2)
# AI040 题面示例 1 / 2；只在前 k 里尝试（4 个词元都想要专家 0、1，容量 1）；c=0 全部丢弃；练习 2 的 c=1
assert moe_route([[9, 5, 1], [8, 7, 2], [1, 2, 9]], 2, 2) == ([[0, 1], [0, 1], [2]], [])
assert moe_route([[5, 5, 5], [5, 5, 5]], 1, 1) == ([[0], []], [1])
assert moe_route([[9, 8, 7]] * 4, 2, 1) == ([[0, 1], [], [], []], [1, 2, 3])
assert moe_route([[1, 2], [3, 4], [5, 6]], 2, 0) == ([[], [], []], [0, 1, 2])
assert moe_route([[9, 5, 1], [8, 7, 2], [1, 2, 9]], 2, 1) == ([[0, 1], [], [2]], [1])见第 07 节展开区（同一份代码）。
练习 5（迁移）：写状态合并函数 merge(a, b) 与区间函数 lse_range(s, l, r)，并验证结合律：对 (1, 1)、(2, 1)、(3, 1) 三个状态，先合前两个再并第三个，与先合后两个再并第一个，m 相同、l 之差小于 10⁻¹²。为什么结合律成立就意味着可以分块、甚至并行归约？
展开练习 5 答案
两种顺序都得到 (3, 1.503215)：l 的每一项最终都是 e^{s_i − 3}，与合并顺序无关；m 取的是 max，也与顺序无关。结合律成立，就可以把长区间切成任意块，每块独立算出 (m, l) 再两两合并——FlashAttention 分块计算的依据正是这一点。AI038 的数据规模用从左到右线性归约即可。
merge 与 lse_range 的参考实现（自带断言）
Pythonimport math
def merge(a, b):
# 状态 (m, l)：m 是区间最大值，l = Σ e^{s_i − m}；l = 0 表示空状态
m1, l1 = a
m2, l2 = b
if l1 == 0.0:
return b
if l2 == 0.0:
return a
m = max(m1, m2)
return m, l1 * math.exp(m1 - m) + l2 * math.exp(m2 - m) # 两边都按新 m 缩放
def lse_range(s, l, r):
# 区间 [l, r]（1 基、闭区间）从左到右逐个合并；答案 = m + ln l
state = (0.0, 0.0)
for i in range(l - 1, r):
state = merge(state, (s[i], 1.0))
m, total = state
return m + math.log(total)
R = lambda v: round(v, 4)
# AI038 题面示例 1 / 2
assert R(lse_range([1.0, 2.0, 3.0], 1, 3)) == 3.4076 and R(lse_range([1.0, 2.0, 3.0], 2, 2)) == 2.0
assert R(lse_range([1e9, -1e9], 1, 2)) == 1e9
# 单元素就是它本身；k 个相同值 = s + ln k
assert R(lse_range([-123.4567], 1, 1)) == -123.4567
assert R(lse_range([2.5] * 8, 1, 8)) == R(2.5 + math.log(8))
# 练习 3：三个数都乘 1000，朴素 Σexp 会上溢，稳定状态给 3000.0000
assert R(lse_range([1000.0, 2000.0, 3000.0], 1, 3)) == 3000.0
# 结合律：先合前两个再并第三个 == 先合后两个再并第一个（分治也能用同一个 merge）
a, b, c = (1.0, 1.0), (2.0, 1.0), (3.0, 1.0)
left, right = merge(merge(a, b), c), merge(a, merge(b, c))
assert left[0] == right[0] and abs(left[1] - right[1]) < 1e-12
# 合并的中间状态：(1, 1) → (2, 1.367879) → (3, 1.503215)
st = merge((1.0, 1.0), (2.0, 1.0))
assert st[0] == 2.0 and abs(st[1] - 1.367879) < 1e-6
st = merge(st, (3.0, 1.0))
assert st[0] == 3.0 and abs(st[1] - 1.503215) < 1e-6见第 07 节展开区（同一份代码）。
10 / 读题要求与复习自评
三道题的要求对照，以及完成本课之后怎么复习
提交前把下表过一遍；题目页的题面与样例是最终依据。
| 项目 | AI041 流水线 | AI040 MoE 路由 | AI038 稳定区间求和 |
|---|---|---|---|
| 输入顺序 | p m；p 个 t；s x | n e k c；n 行 e 个整数 | n q；n 个实数；q 行 l r |
| 编号基准 | 阶段 1 基（s 也是） | 词元、专家都 0 基 | 区间 1 基闭区间 |
| 并列 / 边界 | 瓶颈并列取编号最小；x 可为 0 | 同分取专家编号小；满了跳过不顺延；c 可为 0；k > e → invalid | 单元素答案是它本身；|s| 到 10⁹ |
| 输出 | 两行 总时间 编号 | n 行 t: …/t: - + dropped: …/dropped: - | q 行 4 位小数 |
| 数据范围 | p ≤ 10⁵；m ≤ 10⁹；t ≤ 10⁶（答案约 10¹⁵） | n, e ≤ 200；k ≤ 200；c ≤ 200 | n ≤ 20000；q ≤ 50 |
| 示例 | 3 3 / 2 5 1 / 2 2 → 18 2 / 12 2 | 示例 2 同分 → 0: 0 / 1: - / dropped: 1 | 1 2 3 全区间 → 3.4076 |
需要对照解法时，展开本课第 07 节的三份完整参考程序，或在题目页查看题解。完成条件：AI041、AI040 两道必做题都通过判题，并勾选下方「学习完成检查」的六条，本课才记为完成；进阶练习 AI038 单独统计，不影响完成状态。六条自查是自评，勾选不改变题目的通过（AC）状态。复习时用三个问题自测：① 不看正文，写出总时延公式，并说出 [2, 5, 1]、m=3 的时间线里微批次 2 为什么在时刻 7 才进阶段 2；② 不看表格，重推 AI040 示例 1 第 3 个词元的候选顺序与分配结果；③ 写出 (m, l) 合并公式，说出示例 1 第三次合并的两项各是多少。答不出哪一条，就回到对应的节重读，再做第 09 节对应的练习。
11 / 练习
按顺序完成本课的任务
必做题已通过 0/2 道；进阶练习已通过 0/1 道
AI041 · 推理流水线总耗时与瓶颈
必做任务 1练习重点：总时延公式 + 瓶颈并列取编号最小 + 优化后重新找瓶颈；预计用时：20 分钟
完成标准：能解释公式第二项 (m−1)·max 的含义
需要时查看提示
输出两行「总完成时间 瓶颈编号」。p=1 时变成串行：7 耗时、m=10 → 70。m 达 10⁹ 不能模拟；Python 的整数是任意精度，不会溢出；C++ 用 long long、Java 用 long 才能保证 64 位。第 04 节有示例的时间线表。
AI040 · 混合专家层的容量路由
必做任务 2练习重点：词元按编号顺序占容量；候选只看前 k；满了跳过不顺延；全跳过则丢弃；预计用时：35 分钟
完成标准：同分用例（样例 2）能不看题目说明手工推演出 dropped: 1
需要时查看提示
k > e 直接输出 invalid。输出格式逐字核对：每行「t: 专家列表」，未分配到专家时输出 t: -；最后一行以 dropped: 开头，后接被丢弃编号，没有则输出 dropped: -。c=0 时所有词元都被丢弃，前 n 行仍要逐行输出。第 06 节有示例 1 的逐词元表。
AI038 · 注意力分数的稳定区间求和
进阶练习 1进阶练习练习重点：维护 (m, l) 可结合状态回答区间 LSE 查询；±10⁹ 不溢出；预计用时：20 分钟
完成标准：能说出为什么合并时两边都要按新 m 缩放
需要时查看提示
初始状态 m=−inf、l=0，首次合并要特殊处理；合并时两边都按新 m 重缩放。答案 = m + ln(l)，保留 4 位小数——题目保证答案离舍入边界至少 1e-7，从左到右线性归约即可，不需要分治。先用 LSE(1,2,3)=3.4076 验证，再跑 ±10⁹ 样例。第 06 节有示例 1 的逐次合并表。
提交结果
提交结果说明与处理方法
- WA
答案错误
三个常见错误：优化后沿用旧瓶颈编号、MoE 容量满时向第 k 名之后顺延、同分没按编号小的优先。第 08 节的表给出了每种错误在题库用例上的输出
- PE
格式错误
AI040 的
t: -与dropped: -严格按题目规定的空格、冒号和短横线输出；冒号后有一个空格- RE
运行错误
AI040 读到 k > e 时提前输出
invalid后要正常退出，不要再继续读取矩阵- TLE
超时
AI041 对 m 逐批模拟无法满足数据范围（m 到 10⁹）；AI038 每查询线性归约即可，不要每块重算全量
- AC
通过
分别比较瓶颈选择、专家路由和 NMS 的排序目标及并列规则
12 / 学习完成检查
本课学习完成检查
完成本课需要：必做题全部通过，并勾选本课的全部学习完成检查；进阶练习、基础加练与复习题单独统计，不影响完成状态。登录后，勾选记录会保存到账号，并更新课程总览的完成状态。