P5916. 轻量化推理引擎中的 Masked Fused RMSNorm 算子
场次:2026年10月14日-AI方向 · 来源:CodeFun2000 P5916
题目描述
在轻量化大模型推理引擎中,Transformer 的隐藏状态会在多个子模块之间不断传递。为减少内存读写、提高推理效率,工程实现里常把若干操作并进同一个算子,例如:
- 残差连接 Residual Add
- RMSNorm 归一化
- 逐元素缩放 Scale
- 逐元素偏置 Bias
- Padding Mask 处理
请按下列步骤,实现一个简化的 Masked Fused RMSNorm。
计算步骤
对第 个样本,输入向量记为 ,同时给出残差向量 、残差系数 、有效长度 、缩放向量 、偏置向量 和数值稳定项 。
1. 残差融合
2. 有效部分的 RMS 与 energy
只有前 个位置是有效位置,其后一律视为 padding,不参与 RMS 与 energy。
其中 。
3. 归一化输出
输出向量 按位置取值:
- 有效位置():
- 无效位置():
任务说明
给定 个样本。对每个样本输出三项:
- Masked Fused RMSNorm 之后的结果向量
- 该样本有效部分的
- 该样本有效部分的平方和
输入格式
第一行三个数 :样本个数、每个向量的最大长度、RMSNorm 的数值稳定项。 第二行 个浮点数,依次为缩放向量 。 第三行 个浮点数,依次为偏置向量 。 随后 行,每行一个样本,格式为: 其中:
- 是当前样本的有效长度
- 是残差系数
- 是输入向量
- 是残差向量
数据范围
输出格式
共 行,每行对应一个样本:
要求:
- 相邻数据之间只用一个空格,
|左右也各有一个空格; - 全部浮点数保留两位小数,并按标准四舍五入;
- 不要输出
-0.00。绝对值小于0.005的数一律写成0.00。
示例
示例 1
输入:
2 3 0.5
1.0 1.0 1.0
0.0 2.0 -1.0
3 1.0 2.0 1.0 2.0 1.0 1.0 0.0
2 1.0 0.0 0.0 9.0 0.0 0.0 8.0
输出:
1.21 2.81 -0.19 | 2.48 | 17.00
0.00 2.00 0.00 | 0.71 | 0.00
说明:
第 1 个样本中,,,残差融合后
。
因此
,
,
保留两位小数为 。
再按 计算,三个分量约是 、、,保留两位小数后为
1.21 2.81 -0.19。
第 2 个样本中,,。前两个位置融合后都是 ,第三个位置属于 padding,不计入 energy。
于是 ,
,保留两位小数为 。
有效位置上 ,输出等于对应的 ;无效位置输出 。该行为
0.00 2.00 0.00。
示例 2
输入:
2 3 0.000004
2.0 1.0 0.5
1.0 0.0 -1.0
3 0.0 2.0 2.0 4.0 1.0 3.0 3.0
0 1.0 6.0 6.0 6.0 2.0 2.0 2.0
输出:
2.41 0.71 -0.29 | 2.83 | 24.00
0.00 0.00 0.00 | 0.00 | 0.00
说明:
第 1 个样本的残差系数是 ,残差向量不进入融合:
,
,
,
保留两位小数为 。
代入缩放与偏置后,三个分量约是 、、,保留两位小数为
2.41 0.71 -0.29。
第 2 个样本 ,没有有效元素,因此 , 。 保留两位小数后为 ,输出向量也全部是 。
提示
四舍五入
- 统一保留两位小数,不足两位要补零;
- 不要输出
-0.00,应写成0.00。
输出格式
- 每行依次是 个结果、一个
|、、一个|、; - 空格和换行按上面的格式书写,不要多输出字符;
- 、 按下标 取,不要和样本编号 混用。
解题思路
第一步:理解问题本质
题面把五个算子融合在一起,但对每个样本来说,它仍然只是一次逐元素的数值计算。把公式拆开看,真正需要注意的只有四件事:
-
残差融合:。这里 是第 个样本自己的系数,同一个样本的所有位置共用它。
-
mask 的位置: 决定前多少个位置是有效位置。有效位置参与 energy 统计,也参与归一化输出;其后的位置是 padding。
-
energy 与 RMS 只统计有效部分:
注意 是在除法之后、开方之前加进去的,它保证分母不会恰为 。
-
输出分两段:有效位置输出 ,padding 位置直接输出 。
还有一个容易忽略的对应关系:、 的下标是向量位置 ,不是样本编号 。所有样本共用同一组缩放、偏置向量。
由于样本之间完全独立,逐个样本处理即可,不需要把所有样本同时保存在内存里。
第二步:暴力解法
最直白的写法是严格照着公式翻译:算第 个输出分量时,现场把 重新累加一遍。
for k in range(m):
l, alpha = ...
x, r = ...
h = [x[i] + alpha * r[i] for i in range(n)]
for i in range(n):
if i < l:
# 每算一个位置,就从头再累加一次 energy
energy = 0.0
for j in range(l):
energy += h[j] * h[j]
rms = math.sqrt(energy / max(l, 1) + eps)
y_i = gamma[i] * h[i] / rms + beta[i]
else:
y_i = 0.0
这段代码的结果是对的,但 与 对同一个样本来说是常量:它们只和 的前 项有关,和当前算的是第几个位置无关。把它放进位置循环里重算,复杂度就变成了 。本题 、,最坏约 次乘加,虽然还能跑过,但这些计算全是重复的。
第三步:优化解法——先算 energy,再算输出
把「每个样本变化的量」和「每个位置变化的量」分开:
- 与位置无关:、、;
- 与位置有关:。
于是每个样本只需要两遍扫描:
# 第一遍:融合 + 累加 energy(只累加前 l 个位置)
h = [0.0] * n
energy = 0.0
for i in range(l):
h[i] = x[i] + alpha * r[i]
energy += h[i] * h[i]
# 样本级别的常量只算一次
rms = math.sqrt(energy / max(l, 1) + eps)
# 第二遍:按位置写输出,padding 直接写 0
for i in range(n):
if i < l:
values[i] = gamma[i] * h[i] / rms + beta[i]
else:
values[i] = 0.0
复杂度降到 ,并且 只计算一次。
第四步:最优解法 / 完整实现细节
到第三步为止,算法的复杂度已经是最优的了,剩下的工作都在「实现细节」上,而这道题的判分恰恰卡在细节:
- 一次性读入:用
sys.stdin.buffer.read().split()把全部 token 读进来,再用游标顺序取数。浮点数直接用float(token)解析,行内空格、换行都不会影响。 - 融合与累加合并:对 的循环里同时算 和累加 ;对 再补一遍 (padding 的 不参与输出,但保留完整的 能让「哪些位置被 mask」更直观)。
- 分母取 :当 时没有有效元素,,此时 ,仍然要正常输出,不能除以 。
- padding 输出 而不是 :mask 的语义是「这些位置不产生贡献」,所以偏置也不能加。
- 统一格式化:所有浮点数(包括 与 )都保留两位小数。
f"{v:.2f}"会自动补零,例如 输出17.00。 - 避免
-0.00:浮点误差会让「数学上是 」的结果带一个负号(如-1e-17)。题面规定绝对值小于 的数一律写成0.00,所以格式化前先做一次就近归零,格式化后再检查一次字符串,双保险。 - 退化情况:约束允许 。若此时有效部分又恰好全为 ,则 , 成为 。由于有效位置上 全是 ,按极限把它视为 ,输出退化为 ,代码不会抛
ZeroDivisionError。
完整代码实现
import math
import sys
def format_number(value):
"""保留两位小数,并保证不会输出 "-0.00"。"""
# 题面规定:绝对值小于 0.005 的数一律写成 0.00
if abs(value) < 0.005:
return "0.00"
text = f"{value:.2f}"
# 双保险:极端情况下舍入后仍可能拼出 "-0.00"
return "0.00" if text == "-0.00" else text
def solve():
tokens = sys.stdin.buffer.read().split()
if not tokens:
return
pos = 0
m = int(tokens[pos])
n = int(tokens[pos + 1])
eps = float(tokens[pos + 2])
pos += 3
# gamma、beta 按位置取,所有样本共用,只需要读一次
gamma = [float(tokens[pos + i]) for i in range(n)]
pos += n
beta = [float(tokens[pos + i]) for i in range(n)]
pos += n
lines = []
for _ in range(m):
# 每个样本:l、alpha、x 向量、r 向量
length = int(tokens[pos])
alpha = float(tokens[pos + 1])
pos += 2
if length < 0:
length = 0
elif length > n:
length = n
x = [float(tokens[pos + i]) for i in range(n)]
pos += n
r = [float(tokens[pos + i]) for i in range(n)]
pos += n
# 融合与 energy 合并到同一轮:energy 只累加有效位置
h = [0.0] * n
energy = 0.0
for i in range(length):
h[i] = x[i] + alpha * r[i]
energy += h[i] * h[i]
# padding 位置也补上 h,虽然它不参与输出
for i in range(length, n):
h[i] = x[i] + alpha * r[i]
# 分母取 max(l, 1):l = 0 时 RMS 仍为 sqrt(eps)
divisor = length if length > 0 else 1
rms = math.sqrt(energy / divisor + eps)
values = []
for i in range(n):
if i < length:
if rms == 0.0:
# eps = 0 且有效部分全为 0:把 h/RMS 视为 0
values.append(beta[i])
else:
values.append(gamma[i] * h[i] / rms + beta[i])
else:
# padding 位置输出 0,注意不是 beta[i]
values.append(0.0)
parts = [format_number(value) for value in values]
parts.append("|")
parts.append(format_number(rms))
parts.append("|")
parts.append(format_number(energy))
lines.append(" ".join(parts))
sys.stdout.write("\n".join(lines) + "\n")
if __name__ == "__main__":
solve()
NumPy 版本
以下版本使用 NumPy。只有在评测环境允许第三方库时才能提交;CodeFun2000 的普通 Python 环境不一定安装 NumPy。
import sys
import numpy as np
def format_number(value):
"""保留两位小数,并保证不会输出 "-0.00"。"""
# 题面规定:绝对值小于 0.005 的数一律写成 0.00
if abs(value) < 0.005:
return "0.00"
text = f"{value:.2f}"
# 双保险:极端情况下舍入后仍可能拼出 "-0.00"
return "0.00" if text == "-0.00" else text
def solve():
tokens = sys.stdin.buffer.read().split()
if not tokens:
return
m = int(tokens[0])
n = int(tokens[1])
eps = float(tokens[2])
pos = 3
# gamma、beta 按位置取,所有样本共用,各读成一条长度为 n 的向量
gamma = np.array(tokens[pos:pos + n], dtype=np.float64)
pos += n
beta = np.array(tokens[pos:pos + n], dtype=np.float64)
pos += n
# 每个样本一行,共 2 + 2n 个数:l、alpha、x_1..x_n、r_1..r_n
width = 2 + 2 * n
data = np.array(tokens[pos:pos + m * width], dtype=np.float64).reshape(m, width)
lengths = np.clip(data[:, 0].astype(np.int64), 0, n) # (m,),题面保证 0 <= l <= n
alpha = data[:, 1:2] # (m, 1),按行广播
x = data[:, 2:2 + n] # (m, n)
r = data[:, 2 + n:2 + 2 * n] # (m, n)
# 第 1 步:残差融合 h = x + alpha * r,alpha 逐样本广播到整行
h = x + alpha * r
# 第 2 步:有效位置掩码,第 i 列有效 <=> i < l_k。布尔矩阵直接当 mask
mask = np.arange(n)[None, :] < lengths[:, None]
# 第 3 步:energy 只累加有效位置的平方,padding 位置先清零再平方
# 用 cumsum 取最后一列:它是按列顺序逐项累加的,与「逐位置累加」的
# 结果逐位相同;np.sum 的成对求和虽然更准,但会在 1e22 量级上与逐项
# 累加差出一个 ULP,而 energy 要原样打印两位小数
squares = np.where(mask, h * h, 0.0)
energy = np.cumsum(squares, axis=1)[:, -1]
# 第 4 步:RMS = sqrt(energy / max(l, 1) + eps),eps 在开方里面
divisor = np.maximum(lengths, 1).astype(np.float64)
rms = np.sqrt(energy / divisor + eps)
# 第 5 步:有效位置输出 gamma_i * h_i / RMS + beta_i,padding 位置输出 0
# rms 为 0(eps = 0 且有效部分全为 0)时分子也全为 0,除法按 1 处理即可
safe_rms = np.where(rms > 0.0, rms, 1.0)
y = np.where(
mask,
gamma[None, :] * h / safe_rms[:, None] + beta[None, :],
0.0,
)
lines = []
for k in range(m):
parts = [format_number(value) for value in y[k]]
parts.append("|")
parts.append(format_number(rms[k]))
parts.append("|")
parts.append(format_number(energy[k]))
lines.append(" ".join(parts))
sys.stdout.write("\n".join(lines) + "\n")
if __name__ == "__main__":
solve()
这一版把哪些循环换成了数组运算:
- 残差融合 用
alpha[:, None]把每个样本自己的系数广播到整行, 次乘加一次完成; - 掩码用
np.arange(n)[None, :] < lengths[:, None]直接生成 的布尔矩阵,不需要逐位置判断i < l; energy、、 都是整块算,gamma、beta沿列广播、 沿行广播,逐元素公式与纯 Python 版一字不差;energy这类「逐项累加出来、又要原样打印」的量用np.cumsum(...)[:, -1]求而不是np.sum:np.sum的成对求和更精确,但求和顺序与逐项累加不同,在 达到 量级时会差出一个 ULP,而energy要求原样保留两位小数;- 只有最终格式化仍然逐个取值,因为
f"{v:.2f}"的舍入和-0.00归零按数值判断,保持与纯 Python 版完全一致。
示例推演
样例 1
、、,,。
第 1 个样本:、,,。
先做残差融合:
有效长度等于 ,energy 统计全部三个位置:
,于是
逐分量代入:
| 位置 | 输出 | ||||
|---|---|---|---|---|---|
| 1 | 3 | 0 | 1.2081 | 1.21 | |
| 2 | 2 | 2 | 2.8054 | 2.81 | |
| 3 | 2 | -0.19 |
该行输出 1.21 2.81 -0.19 | 2.48 | 17.00。
第 2 个样本:、,,。
融合后 。只有前 个位置有效,而它们的 都是 ,所以 。:
位置 1、2 上 ,输出就是偏置本身:、;位置 3 是 padding,输出 。
该行输出 0.00 2.00 0.00 | 0.71 | 0.00。
样例 2
、、,,。
第 1 个样本:、,,。
时残差项整体为 ,所以 。
| 位置 | 输出 | |||
|---|---|---|---|---|
| 1 | 1 | 2.4142 | 2.41 | |
| 2 | 0 | 0.7071 | 0.71 | |
| 3 | -0.29 |
该行输出 2.41 0.71 -0.29 | 2.83 | 24.00。
第 2 个样本:、,,。
表示没有任何有效位置:,,
三个位置全是 padding,全部输出 。该行输出 0.00 0.00 0.00 | 0.00 | 0.00。
注意这里 明明不为零,但因为全部被 mask,既不进 也不进输出。这正是 Masked 的语义。
复杂度分析
| 解法 | 时间复杂度 | 空间复杂度 | 说明 |
|---|---|---|---|
| 暴力:逐位置重算 energy | 每个输出位置都从头累加一次平方和 | ||
| 优化:energy、RMS 只算一次 | 每个样本两遍扫描 | ||
| 本题实现(融合 + 单次累加) | 融合与累加合并成一轮,常数更小 |
、 时, 只有约 次浮点乘加,远远低于 2 秒的限制。
易错点总结
1. energy 只统计前 个位置
padding 位置的 即使很大也不能进 。样例 1 第 2 个样本的 ,如果误把它算进去, 会变成 , 与整行输出全错。
2. 分母是 ,不是
时直接写 energy / l 会抛出 ZeroDivisionError。题目已经给出 ,照抄即可。
3. 在开方里面
正确写法是 ,不是 。把 提到根号外,数值会明显不同。
4. padding 输出 ,不是
mask 之后这些位置不产生贡献,缩放与偏置都不该加。样例 1 第 2 个样本的位置 3 若输出 ,就会得到 -1.00 而不是 0.00。
5. 、 用位置下标,不用样本编号
所有样本共用同一组 、。写成 gamma[k] 在 时直接越界,在 时会静默取错值。
6. 不要输出 -0.00
浮点运算会出现 -1e-17 这类「数学上是 0」的结果。题面明确要求绝对值小于 的数写成 0.00,所以格式化前必须先归零。
7. 保留两位小数要补零
17 要输出 17.00,0 要输出 0.00。使用 f"{value:.2f}" 可以自动补齐;round(value, 2) 之后再 str() 不行,因为 str(round(17.0, 2)) 得到的是 17.0,不会补成 17.00。
8. 的退化情形
约束允许 。如果有效位置上的 全为 ,则 ,归一化项是 。代码里把这种情况视为 ,输出退化为 ,既保证不崩溃,也与「有效位置没有信号」的直觉一致。
9. 输出分隔符
每行是 个结果、|、、|、。用 " ".join(...) 拼接可以保证相邻数据之间只有一个空格,| 左右各一个空格。
扩展思考
- 真正的融合算子:工程实现里这一步通常写成 kernel,残差写回、RMS 统计、缩放偏置一趟走完,避免把中间张量写回显存。本题的「先算 energy 再算输出」正是这种单遍思路的简化版。
- 数值稳定性: 最大 、 最大 ,融合后 可达 , 可达 。用双精度(float64)累加完全够用;如果改用 float32,累加 256 项时会损失有效精度,通常的做法是先把每个平方项按块累加,或者用补偿求和(Kahan summation)。
- 分组 RMSNorm:真实模型里常把隐藏维度分成若干组,每组单独算 RMS。做法与本题完全一样,只是把「前 个位置」换成「某一组的下标区间」。
- 两种 mask 的区别:本题是 padding mask(按有效长度截断),自回归模型还需要 causal mask(第 个位置只能看到 )。两者可以叠加,此时 的累加上界要取两者的交集。
- 多 batch 的写法:本题样本之间互不影响,所以逐样本处理即可;如果输入改成「先给一个矩阵再给一个 mask 矩阵」,也可以先算每一行的 ,再整块广播计算 ,结果相同。