Skip to content
Gains Summary
Main Navigation 首页 / Home
C++ 编程 / C++ Programming
系统与高性能 / Systems & Performance
Web 开发 / Web Development
人工智能 / Artificial Intelligence
工业软件 / Industrial Software
其他内容 / Other Topics
C++ 编程 / C++系统与性能 / SystemsWeb 开发 / Web人工智能 / AI工业软件 / Industrial

外观

Sidebar Navigation

← 人工智能 / Artificial Intelligence

AI 编译器 / AI Compilers

1. AI 编译器全景——为什么模型需要编译器 / The AI Compiler Landscape and Why Models Need Compilers

2. 编译原理速通——面向 ML 工程师的核心概念 / Compiler Fundamentals for Machine Learning Engineers

3. 中间表示基础——理解 IR 层级与 lowering 链路 / Intermediate Representation Levels and Lowering Pipelines

4. 计算图的构建与表示 / Building and Representing Computational Graphs

5. MLIR 架构、方言与渐进式降级 / MLIR Architecture, Dialects, and Progressive Lowering

6. 算子语义、广播、归约与形状推导 / Operator Semantics, Broadcasting, Reduction, and Shape Inference

7. 模型前端格式:ONNX、TFLite、HLO 与 SavedModel / Model Frontend Formats: ONNX, TFLite, HLO, and SavedModel

8. 图优化 Pass——经典优化在 ML 中的应用 / Graph Optimization Passes for Machine Learning

9. 算子融合——编译器最重要的性能优化 / Operator Fusion as a Core Compiler Optimization

10. 内存规划——Buffer 分配与生命周期管理 / Memory Planning, Buffer Allocation, and Lifetime Management

11. Layout 优化——数据排布转换与内存效率 / Layout Optimization for Data Movement and Memory Efficiency

12. 动态 Shape——符号分析与形状处理 / Dynamic Shapes, Symbolic Analysis, and Shape Processing

13. 硬件约束下的操作调度 / Operation Scheduling Under Hardware Constraints

14. 从模板、DSL 到 IR 降级的代码生成架构 / Code Generation Architectures from Templates and DSLs to IR Lowering

15. CPU 后端:SIMD、分块与多线程 / CPU Backends with SIMD, Tiling, and Multithreading

16. CUDA 后端:合并访存与 Tensor Core / CUDA Backends, Memory Coalescing, and Tensor Cores

17. NPU 后端:脉动阵列与端侧 AI 生态 / NPU Backends, Systolic Arrays, and Edge AI Ecosystems

18. Kernel 性能基础:Roofline 与 Occupancy / Kernel Performance Fundamentals with Roofline and Occupancy

19. CUTLASS 与分层 GEMM 模板 / CUTLASS and Hierarchical GEMM Templates

20. TVM Tensor Expression 与计算调度分离 / TVM Tensor Expressions and Compute-Schedule Separation

21. 使用 Triton 编写高性能 GPU Kernel / Triton for High-Performance GPU Kernels in Python

22. 基于成本模型与实测搜索的自动调度 / Automatic Scheduling with Cost Models and Measurement-Based Search

23. XLA 内部机制:HLO、融合与 SPMD / XLA Internals, HLO, Fusion, and SPMD

24. Torch-MLIR:从 PyTorch 算子到 MLIR 方言 / Torch-MLIR from PyTorch Operators to MLIR Dialects

25. torch.compile:Dynamo、AOTAutograd、Inductor 与 Triton / Torch Compile with Dynamo, AOTAutograd, Inductor, and Triton

26. 从 MLIR 经 LLVM 降级到机器码 / Lowering from MLIR Through LLVM to Machine Code

27. 量化——低精度推理的工程实践 / Engineering Low-Precision Inference with Quantization

28. 分布式编译与训练——多设备编排的编译器支持 / Compiler Support for Distributed Training and Multi-Device Orchestration

29. 生产调试——真实问题的编译器视角排查 / Production Debugging from the Compiler Perspective

30. 未来方向——AI 编译器的新挑战与机遇 / Future Challenges and Opportunities for AI Compilers

本页目录

📅 创建时间:2026-06-03 🏷️ 标签:#图优化 #Pass #CSE #DCE #ConstantFolding #AlgebraicSimplification #代简简化 #死代码消除 📚 前置知识:[[/04-ai/01-llm-engineering/07-llm-evolution]](LLM 发展脉络) 📚 相关知识:[[03-graph-representation]](计算图表示) 📚 相关知识:[[08-operator-fusion]](算子融合) [[09-memory-planning]](内存规划)


┌──────────────────────────────────────────────────────────────────────────────┐ │ 🔍 场景:torch.export 导出的模型为什么这么大? │ ├──────────────────────────────────────────────────────────────────────────────┤ │ 你用 torch.export 导出了一个模型,检查导出的 graph: │ │ - 原始模型:500MB 的权重文件 │ │ - 导出的 torch.export 模型:3GB 的完整计算图文件 │ │ - 你的困惑:权重才 500MB,怎么导出来变成 3GB 了? │ │ - 编译器是不是把整个 Python 运行时都塞进去了? │ └──────────────────────────────────────────────────────────────────────────────┘

图优化 Pass——经典优化在 ML 中的应用 / Graph Optimization Passes for Machine Learning ​

第1节:什么是 Pass——分析遍与变换遍 ​

在传统编译器(如 GCC、LLVM)中,Pass 是编译器处理代码的基本单元。每个 Pass 负责一项特定的任务:要么分析代码(分析遍),要么转换代码(变换遍)。ML 编译器继承了这一概念,将其应用于计算图。

1.1 分析遍(Analysis Pass) ​

分析遍不修改计算图,而是收集信息供其他 Pass 使用:

python
# 分析遍的典型输出
class ShapeAnalysis:
    """
    分析遍示例:收集图中每个算子的输出形状信息
    """
    
    def run(self, graph):
        shape_map = {}
        
        for node in graph.nodes:
            # 根据输入形状和算子类型推导输出形状
            if node.op_type == "conv2d":
                # Conv2D 输出形状计算公式
                # output_H = (input_H - kernel_H + 2*pad) / stride + 1
                # output_W = (input_W - kernel_W + 2*pad) / stride + 1
                shape_map[node.name] = self._infer_conv_shape(node)
                
            elif node.op_type == "matmul":
                # MatMul: (M, K) × (K, N) → (M, N)
                shape_map[node.name] = self._infer_matmul_shape(node)
        
        return shape_map
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

1.2 变换遍(Transformation Pass) ​

变换遍基于分析结果修改计算图:

python
# 变换遍示例:常量折叠
class ConstantFolding:
    """
    变换遍示例:将编译期可计算的常量表达式直接求值
    """
    
    def run(self, graph):
        modified = True
        while modified:
            modified = False
            
            for node in graph.nodes:
                # 检查是否所有输入都是常量
                if all(inp.is_constant() for inp in node.inputs):
                    # 替换为计算结果
                    result = self._evaluate(node)
                    graph.replace(node, constant_node(result))
                    modified = True
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

1.3 Pass 的组合方式 ​

现代 ML 编译器通常将多个 Pass 组合成 Pipeline:

python
# Pass Pipeline 示例
class CompilerPipeline:
    def __init__(self):
        self.passes = [
            # 阶段1:规范化
            CanonicalizationPass(),
            
            # 阶段2:常量相关优化
            ConstantFoldingPass(),
            ConstantPropagationPass(),
            
            # 阶段3:死代码消除
            DeadCodeEliminationPass(),
            
            # 阶段4:算子融合
            FuseOpsPass(),
            
            # 阶段5:内存优化
            MemoryPlanningPass(),
        ]
    
    def run(self, graph):
        for pass_obj in self.passes:
            print(f"Running {pass_obj.name}...")
            graph = pass_obj.run(graph)
        return graph
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

第2节:经典编译优化在 ML 图中的应用 ​

2.1 DCE(Dead Code Elimination)——死代码消除 ​

DCE 是最重要的优化之一,它消除计算图中不会被使用的算子和常量。

python
# DCE 算法实现
class DeadCodeElimination:
    """
    死代码消除:删除所有不影响程序输出的代码
    
    算法步骤:
    1. 从输出节点反向标记所有活着的节点(Liveness Analysis)
    2. 删除所有未被标记的节点
    """
    
    def run(self, graph):
        # Step 1: 找出所有输出(loss, 推理结果等)
        outputs = self._find_outputs(graph)
        
        # Step 2: 反向传播,标记活着的节点
        alive = set()
        worklist = list(outputs)
        
        while worklist:
            node = worklist.pop()
            if node in alive:
                continue
            
            alive.add(node)
            
            # 所有输入这个节点的节点也要活着
            for inp in node.inputs:
                if inp not in alive:
                    worklist.append(inp)
        
        # Step 3: 删除不在 alive 集合中的节点
        for node in list(graph.nodes):
            if node not in alive:
                print(f"Removing dead node: {node.name}")
                graph.remove(node)
        
        return graph
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

DCE 在 ML 中的典型应用场景:

场景原始图DCE 后节省
丢弃的 auxiliary loss额外的 softmax+cross_entropy删除整个子树~200MB
推理时的 dropoutDropout mask 生成删除(推理时为 identity)~50MB
未使用的 tower多任务学习的辅助 tower删除整个分支~100MB
调试用的 print多次中间结果打印删除所有打印节点~10MB

2.2 CSE(Common Subexpression Elimination)——公共子表达式消除 ​

CSE 消除重复的相同计算:

python
# CSE 实现
class CommonSubexpressionElimination:
    """
    公共子表达式消除:如果两个表达式计算出相同结果,保留一个
    
    示例:
    y1 = add(mul(a, b), mul(a, b))  →  t = mul(a, b); y1 = add(t, t)
    """
    
    def run(self, graph):
        # 使用哈希表存储已见过的表达式
        seen_expressions = {}
        
        for node in list(graph.nodes):
            # 生成表达式的规范表示
            expr_key = self._canonicalize(node)
            
            if expr_key in seen_expressions:
                # 发现重复表达式,用已存在的替换
                existing = seen_expressions[expr_key]
                print(f"CSE: {node.name} → {existing.name}")
                graph.replace_all_uses(node, existing)
                graph.remove(node)
            else:
                seen_expressions[expr_key] = node
        
        return graph
    
    def _canonicalize(self, node):
        # 生成唯一的表达式键
        return (
            node.op_type,
            tuple(id(inp) for inp in node.inputs)
        )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

ML 中的 CSE 示例:

python
# 原始代码
q = linear(x, w_q)  # Q = W_q @ x
k = linear(x, w_k)  # K = W_k @ x
v = linear(x, w_v)  # V = W_v @ x

# CSE 可以发现 x 被重复使用,但这里的 x 每次都是同一个输入
# CSE 不会改变上面的代码,因为 op_type 不同

# 但如果是这样:
tmp = relu(linear(x, w1))
y1 = linear(tmp, w2)
y2 = linear(tmp, w3)

# CSE 可以复用 tmp,避免重复计算 relu(linear(x, w1))
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.3 Constant Folding——常量折叠 ​

常量折叠在编译期计算常量表达式:

python
# 常量折叠实现
class ConstantFolding:
    """
    常量折叠:将编译期可确定的常量表达式直接求值
    
    示例:
    BatchNorm 的 mean 和 variance 在训练后是固定的
    可以预先计算好 gamma / sqrt(variance + epsilon) * x + (beta - gamma*mean/sqrt(variance+eps))
    融合成单个 affine 算子
    """
    
    def fold_batchnorm(self, bn_node, graph):
        """
        BatchNorm 参数融合示例
        
        BatchNorm 计算公式:
        y = (x - mean) / sqrt(var + eps) * gamma + beta
        
        融合后变成:
        y = gamma / sqrt(var + eps) * x + (beta - gamma * mean / sqrt(var + eps))
        即:y = weight * x + bias
        """
        gamma = bn_node.inputs[1].value  # scale
        beta = bn_node.inputs[2].value   # bias
        mean = bn_node.inputs[3].value   # running_mean
        var = bn_node.inputs[4].value    # running_var
        eps = bn_node.attrs.get('eps', 1e-5)
        
        # 计算融合后的权重和偏置
        scale = gamma / np.sqrt(var + eps)
        bias = beta - gamma * mean / np.sqrt(var + eps)
        
        # 创建融合后的 Affine 节点
        fused = graph.create_node(
            op_type='affine',
            inputs=[bn_node.inputs[0]],  # 只需要输入 x
            attrs={'weight': scale, 'bias': bias}
        )
        
        graph.replace(bn_node, fused)
        print(f"Fused BatchNorm into Affine: saved {len(bn_node.inputs)} inputs")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

常量折叠在 ML 中的典型应用:

优化前优化后收益
reshape([-1, 1]) + squeeze()保持原 shape避免中间张量
matmul(const_A, const_B)const_C = matmul(A, B)运行时省一次矩阵乘法
conv(weight) + add(bias)单个带 bias 的 conv减少 kernel 调用

2.4 Algebraic Simplification——代数简化 ​

代数简化利用数学恒等式消除冗余操作:

python
# 代数简化规则
class AlgebraicSimplification:
    """
    代数简化规则表
    
    规则1: 1 × x = x
    规则2: 0 + x = x
    规则3: x / x = 1 (x ≠ 0)
    规则4: x ^ 0 = 1
    规则5: x ^ 1 = x
    规则6: abs(abs(x)) = abs(x)
    规则7: relu(relu(x)) = relu(x)
    规则8: reshape(reshape(x, ...), ...) = reshape(x, ...)
    """
    
    RULES = [
        # (pattern, replacement, condition)
        ('mul', [('const', 1)], 'identity'),  # x * 1 = x
        ('mul', [('const', 0)], 'zero'),       # x * 0 = 0
        ('add', [('const', 0)], 'identity'),  # x + 0 = x
        ('sub', [('const', 0)], 'identity'),  # x - 0 = x
        ('div', [('const', 1)], 'identity'),   # x / 1 = x
        ('pow', [('const', 0)], 'one'),        # x ^ 0 = 1
        ('pow', [('const', 1)], 'identity'),  # x ^ 1 = x
        ('relu', ['relu'], 'nested_relu'),    # relu(relu(x)) = relu(x)
    ]
    
    def run(self, graph):
        for node in graph.nodes:
            for pattern, replacement, condition in self.RULES:
                if self._match(node, pattern, condition):
                    graph.replace(node, self._create_simplified(node, replacement))
                    break
        return graph
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

ML 中 1×1 Conv 的消除:

python
# 1×1 Conv 是代数简化的典型场景
# 1×1 Conv 实际上就是全连接操作

# 原始
x → conv(1x1, stride=1, pad=0) → y

# 对于每个空间位置:
# y[h,w,c] = Σ_k x[h,w,k] * weight[k,c]

# 这完全等价于:
# y = x @ weight  (矩阵乘法)

# 所以可以简化为:
x → linear(in_features=K, out_features=C) → y

# 消除的好处:
# - 1x1 conv 有额外的 im2col 开销
# - linear/matmul 有更高效的 kernel(如 cuBLAS)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

2.5 Canonicalization——规范化 ​

规范化将等价但形式不同的算子转换成统一形式:

python
# 规范化示例
class Canonicalization:
    """
    规范化:将图转换成标准形式
    
    示例:
    - add(x, y) 和 add(y, x) 统一为 add(x, y)
    - sub(x, 0) 统一为 x
    - mul(x, 2) 统一为 add(x, x)
    - reshape(x, [-1]) 统一为 flatten(x)
    """
    
    def canonicalize_add(self, node, graph):
        # 确保 x + y 中 x 不是常量(常量放右边便于后续折叠)
        if node.inputs[0].is_constant():
            node.inputs = [node.inputs[1], node.inputs[0]]
    
    def canonicalize_reshape(self, node, graph):
        # 合并相邻的 reshape
        if node.inputs[0].op_type == 'reshape':
            prev_reshape = node.inputs[0]
            # reshape(reshape(x, A), B) = reshape(x, B)
            graph.replace_input(node, 0, prev_reshape.inputs[0])
    
    def canonicalize_transpose(self, node, graph):
        # transpose(transpose(x)) = x
        if node.inputs[0].op_type == 'transpose':
            prev = node.inputs[0]
            if self._is_inverse(prev.attrs, node.attrs):
                graph.replace(node, prev.inputs[0])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

第3节:Shape/Dtype 推导——编译期的形状推断 ​

3.1 静态形状推导 ​

python
# 形状推导器
class ShapeInference:
    """
    编译器如何在不运行模型的情况下推断张量形状
    
    原理:根据算子的数学定义,从输入形状推导输出形状
    """
    
    def infer_conv2d(self, input_shape, weight_shape, stride, padding):
        """
        Conv2D 形状推导
        
        输入: (N, C_in, H_in, W_in)
        权重: (C_out, C_in, K_h, K_w)
        输出: (N, C_out, H_out, W_out)
        
        其中:
        H_out = (H_in + 2*pad_h - K_h) // stride + 1
        W_out = (W_in + 2*pad_w - K_w) // stride + 1
        """
        N, C_in, H_in, W_in = input_shape
        C_out, _, K_h, K_w = weight_shape
        pad_h, pad_w = padding
        stride_h, stride_w = stride
        
        H_out = (H_in + 2 * pad_h - K_h) // stride_h + 1
        W_out = (W_in + 2 * pad_w - K_w) // stride_w + 1
        
        return (N, C_out, H_out, W_out)
    
    def infer_matmul(self, a_shape, b_shape):
        """
        MatMul 形状推导
        
        (M, K) @ (K, N) = (M, N)
        """
        assert a_shape[-1] == b_shape[-2], "Matmul dimension mismatch"
        return a_shape[:-1] + (b_shape[-1],)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

3.2 符号形状(Symbolic Shape) ​

python
# 符号形状表示
class SymbolicShape:
    """
    用符号表示不确定的维度
    
    例如:动态 batch size 的输入
    input_shape: (B, 768) 其中 B 是符号
    """
    
    def __init__(self, dims):
        self.dims = dims  # dims 可以是 int 或 Symbol
    
    def __repr__(self):
        return f"SymbolicShape({self.dims})"


class Symbol:
    """符号维度"""
    
    def __init__(self, name, constraints=None):
        self.name = name
        self.constraints = constraints or {}
    
    def __repr__(self):
        return f"${self.name}"


# 示例:动态 Batch
batch_sym = Symbol("B", constraints={"lower": 1, "upper": 128})
seq_len_sym = Symbol("L", constraints={"lower": 1, "upper": 2048})

# 形状表达式
input_shape = SymbolicShape([batch_sym, seq_len_sym, 768])
# 推导 attention output
# attention: (B, L, 768) @ (768, 768) = (B, L, 768)
# output_shape: SymbolicShape([$B, $L, 768])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

3.3 形状传播示例 ​

python
# 完整的形状传播流程
def propagate_shapes(graph, input_shapes):
    """
    从输入开始,依次推导所有节点的输出形状
    """
    shape_map = dict(input_shapes)  # 初始化输入形状
    
    for node in topological_sort(graph):
        input_shapes = [shape_map[inp] for inp in node.inputs]
        
        # 根据算子类型推导输出形状
        if node.op_type == 'linear':
            # linear: (..., in_features) @ (out_features, in_features) → (..., out_features)
            out_features = node.attrs['out_features']
            output_shape = input_shapes[0][:-1] + (out_features,)
            
        elif node.op_type == 'conv2d':
            output_shape = ShapeInference().infer_conv2d(
                input_shapes[0], node.inputs[1].shape,
                node.attrs['stride'], node.attrs['padding']
            )
            
        elif node.op_type == 'reshape':
            # 从 attrs 获取目标 shape
            target_shape = node.attrs['shape']
            output_shape = SymbolicShape(target_shape)
        
        shape_map[node.name] = output_shape
    
    return shape_map
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

第4节:算子融合作为 Pass ​

4.1 融合 Pass 的结构 ​

python
class FuseOpsPass:
    """
    算子融合 Pass:寻找可融合的模式并生成融合后的算子
    
    融合策略:
    1. 贪婪匹配:从输出往输入遍历,找匹配模式
    2. 动态规划:找最优融合方案(NP-hard,近似算法)
    """
    
    FUSION_PATTERNS = [
        # Conv + BN 融合
        {
            'name': 'conv_bn',
            'pattern': ['conv2d', 'batch_norm'],
            'fused_op': 'conv2d_bn',
        },
        # Conv + ReLU 融合
        {
            'name': 'conv_relu',
            'pattern': ['conv2d', 'relu'],
            'fused_op': 'conv2d_relu',
        },
        # MatMul + Add(Bias)融合
        {
            'name': 'linear_bias',
            'pattern': ['matmul', 'add'],
            'fused_op': 'linear',
        },
    ]
    
    def run(self, graph):
        for pattern_def in self.FUSION_PATTERNS:
            self._fuse_pattern(graph, pattern_def)
        return graph
    
    def _fuse_pattern(self, graph, pattern_def):
        pattern = pattern_def['pattern']
        fused_op = pattern_def['fused_op']
        
        # 贪婪匹配
        matched = self._find_pattern_matches(graph, pattern)
        
        for match in matched:
            # 替换为融合算子
            fused_node = graph.create_fused_node(fused_op, match)
            graph.replace(match, fused_node)
            print(f"Fused {pattern} → {fused_op}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

4.2 融合示例:Conv + BN ​

python
# Conv + BatchNorm 融合前后对比

# 融合前:
# conv_out = conv2d(x, weight_conv)
# bn_out = batch_norm(conv_out, gamma, beta, mean, var)
# y = relu(bn_out)

# 融合后:
# weight_fused = gamma / sqrt(var + eps) * weight_conv
# bias_fused = beta - gamma * mean / sqrt(var + eps)
# y = conv2d_relu(x, weight_fused, bias_fused)

# 单个 kernel 替代三个,减少内存访问
1
2
3
4
5
6
7
8
9
10
11
12
13

第5节:Pass 执行顺序与依赖管理 ​

5.1 依赖分析 ​

python
class PassScheduler:
    """
    Pass 调度器:管理 Pass 的执行顺序和依赖
    """
    
    def __init__(self):
        self.dependencies = {
            'DCE': [],
            'CSE': ['Canonicalization'],
            'ConstantFolding': ['Canonicalization'],
            'FuseOps': ['DCE', 'ConstantFolding'],
            'MemoryPlanning': ['FuseOps', 'ShapeInference'],
        }
        
        self.pass_outputs = {
            'DCE': ['liveness_info'],
            'ShapeInference': ['shape_map'],
            'FuseOps': ['fused_graph'],
        }
    
    def get_execution_order(self):
        """
        拓扑排序确定执行顺序
        """
        # 简单的 Kahn 算法
        in_degree = {p: 0 for p in self.dependencies}
        adj_list = {p: [] for p in self.dependencies}
        
        for pass_name, deps in self.dependencies.items():
            for dep in deps:
                adj_list[dep].append(pass_name)
                in_degree[pass_name] += 1
        
        # BFS 拓扑排序
        queue = [p for p, d in in_degree.items() if d == 0]
        order = []
        
        while queue:
            pass_name = queue.pop(0)
            order.append(pass_name)
            
            for next_pass in adj_list[pass_name]:
                in_degree[next_pass] -= 1
                if in_degree[next_pass] == 0:
                    queue.append(next_pass)
        
        return order
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

5.2 常见的 Pass 执行顺序 ​

阶段Pass 顺序原因
规范化Canonicalization → ConstantFolding规范化后常量更容易识别
清理DCE → CSE删除死代码后有更多 CSE 机会
融合FuseOps → DCE融合可能产生新的死代码
内存MemoryPlanning需要所有融合完成后才能规划
python
# 典型的 ML 编译器 Pass Pipeline
STANDARD_PASS_PIPELINE = [
    # 1. 规范化阶段
    ('Canonicalization', '标准化算子形式'),
    
    # 2. 常量优化阶段
    ('ConstantFolding', '折叠常量表达式'),
    ('ConstantPropagation', '传播常量值'),
    ('AlgebraicSimplification', '代数简化'),
    
    # 3. 死代码消除
    ('DCE', '消除无用代码'),
    
    # 4. 子表达式消除
    ('CSE', '消除重复计算'),
    
    # 5. 形状推导(分析遍)
    ('ShapeInference', '推导张量形状'),
    ('DtypeInference', '推导数据类型'),
    
    # 6. 算子融合
    ('ConvBnFusion', '融合 Conv+BN'),
    ('ElementwiseFusion', '融合逐元素算子'),
    ('MatmulFusion', '融合矩阵乘法链'),
    
    # 7. 内存优化
    ('MemoryPlanning', '规划内存分配'),
    ('LayoutOptimization', '优化数据排布'),
    
    # 8. 代码生成准备
    ('OpSpecialization', '算子特化'),
]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

第6节:调试优化效果——查看图差异 ​

6.1 torch.compile 优化查看 ​

python
import torch
from torch._dynamo import config as dynamo_config
from torch.fx import symbolic_trace

# 方法1: 查看优化前的图
model = torch.nn.Sequential(
    torch.nn.Linear(128, 256),
    torch.nn.ReLU(),
    torch.nn.Linear(256, 10),
)

# 使用 torch.compile 捕获优化
compiled_model = torch.compile(model, backend='inductor')

# 查看中间表示
@torch.compile(backend='inductor', fullgraph=True)
def simple_model(x):
    return model(x)

# 运行以触发编译
x = torch.randn(1, 128)
output = simple_model(x)

# 方法2: 导出 ONNX 查看
torch.onnx.export(model, x, "model_before_optimization.onnx")

# 方法3: 使用 FX 跟踪
traced = symbolic_trace(model)
print(traced.graph)
# 输出:
# graph():
#     %x : torch.Tensor
#     %linear1_weight : torch.Tensor
#     %linear1_bias : torch.Tensor
#     %linear2_weight : torch.Tensor
#     %linear2_bias : torch.Tensor
#     %linear1_out : torch.Tensor = call_function[target=torch.nn.functional.linear](args = (%x, %linear1_weight, %linear1_bias))
#     %relu1_out : torch.Tensor = call_function[target=torch.nn.functional.relu](args = (%linear1_out,))
#     %linear2_out : torch.Tensor = call_function[target=torch.nn.functional.linear](args = (%relu1_out, %linear2_weight, %linear2_bias))
#     return linear2_out
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

6.2 生成 .dot 文件可视化 ​

python
# 导出图为 .dot 格式用于 Graphviz 可视化
from torch.fx.passes.graph_drawer import FxGraphDrawer

def export_graph_to_dot(model, filename):
    """
    导出计算图为 .dot 文件
    
    使用方法:
    dot -Tpng model.dot -o model.png
    """
    traced = symbolic_trace(model)
    
    drawer = FxGraphDrawer(traced, filename)
    dot_graph = drawer.get_dot_graph()
    
    with open(f"{filename}.dot", 'w') as f:
        f.write(dot_graph)
    
    print(f"Exported to {filename}.dot")
    print("Convert to PNG: dot -Tpng model.dot -o model.png")


# 优化前后对比
model_before = create_original_model()
model_after = create_optimized_model()

export_graph_to_dot(model_before, "graph_before")
export_graph_to_dot(model_after, "graph_after")

# 使用系统命令生成对比图
import subprocess
subprocess.run(['dot', '-Tpng', 'graph_before.dot', '-o', 'graph_before.png'])
subprocess.run(['dot', '-Tpng', 'graph_after.dot', '-o', 'graph_after.png'])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

6.3 使用 torch._inductor 分析 ​

python
# 分析 Inductor 编译输出
import torch._inductor.config as inductor_config

# 开启调试选项
inductor_config.debug = True
inductor_config.verbose_progress = True

# 编译并查看日志
model = create_transformer_model()
compiled = torch.compile(model, backend='inductor')

# 第一次运行会触发编译
output = compiled(input_tensor)

# 查看生成的代码
print(inductor_config.output_code)
# 会在当前目录生成 torchinductor_my_model/ 目录
# 包含生成的 CUDA/OpenMP 代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

第7节:常见 ML 图优化 Pass 一览表 ​

Pass 名称类型输入输出收益示例
DCE变换完整图精简图减少计算量和内存删除推理时的 Dropout
CSE变换重复计算单一计算减少重复计算多次使用同一子表达式
ConstantFolding变换含常量表达式求值结果运行时省计算BatchNorm 参数融合
AlgebraicSimplify变换冗余操作简化操作减少 kernel 调用消除 1×1 Conv
Canonicalization变换非标准图标准图便于后续优化合并相邻 Reshape
ShapeInference分析部分形状完整形状支持后续优化从输入推导所有形状
FuseOps变换多算子融合算子减少 kernel launchConv+ReLU 融合
MemoryPlanning变换任意图内存优化图减少显存占用Activation 重用

升华 ​

┌─────────────────────────────────────────────────────────────────┐
│                    图优化 Pass 核心原则                           │
├─────────────────────────────────────────────────────────────────┤
│  1. 先分析,后变换                                               │
│     → 形状分析和依赖分析是所有优化的基础                          │
│  2. 规范化先行                                                   │
│     → Canonicalization 使后续所有优化更有效                       │
│  3. 清理优于融合                                                 │
│     → DCE + CSE 在前,融合在后,避免融合死代码                   │
│  4. 融合后重新清理                                               │
│     → 融合可能产生新的死代码,需要再次 DCE                       │
│  5. 顺序敏感                                                     │
│     → Pass 顺序不对可能导致优化失效甚至变慢                       │
└─────────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14

"AI 可查 vs 必须理解"清单 ​

必须理解(不理解就等于不会):

  • 🔴 Pass 的两种类型:分析遍(收集信息)和变换遍(修改图)的区别
  • 🔴 DCE 算法:Liveness Analysis 是如何判断一个节点是否"活着"的
  • 🔴 常量折叠的原理:为什么 BatchNorm 可以被融合成一个 Affine 算子
  • 🔴 Pass 依赖关系:为什么 DCE 要在 CSE 之前执行
  • 🔴 形状推导:从输入形状如何一步步推导所有节点的输出形状

AI 可查(知道去哪查就行):

  • ✅ 特定算子的融合模式(如 Flash Attention 的融合策略)
  • ✅ 某个框架的 Pass Pipeline 配置(如 torch.compile 的优化级别)
  • ✅ .dot 文件的 Graphviz 可视化命令
  • ✅ 特定融合 kernel 的生成代码(如 cuBLAS 的 GEMM 实现)
  • ✅ 某个异常形状推导的处理方法

学习状态:🟡 开始学习

最后更新于:

Pager
上一篇7. 模型前端格式:ONNX、TFLite、HLO 与 SavedModel / Model Frontend Formats: ONNX, TFLite, HLO, and SavedModel
下一篇9. 算子融合——编译器最重要的性能优化 / Operator Fusion as a Core Compiler Optimization

持续记录,持续成长

Copyright © Tidenflow