计算图：数据流图

可视化演示

展示数据流图三维度：DAG结构（节点=算子Conv/BN/Relu/Add、边=张量数据流、拓扑排序确定执行顺序）、静态vs动态图（TF先构建后优化执行 vs PT边构建边执行eager模式）、算子融合（Conv+BN+Relu三算子融合为单个Fused算子减少内存读写），三模式切换理解AI编译器核心中间表示。

1. 图的两大要素

在现代化后厨里，一张流程图只要说清楚两件事：有哪些工序，以及半成品如何流向下一道工序。数据流图同样只关心两件事：节点和边。

节点表示计算动作，也就是算子，比如矩阵乘法、卷积、加法；边表示张量数据在算子之间的流动方向。前端框架把模型翻译成图时，先把每个算子放进节点，再把“谁的结果给谁用”记录成有向边。整张图就是一份“谁等谁、谁喂谁”的流水线说明。

因为数据只能从生产它的算子流到消费它的算子，不能反过来，所以数据流图通常是有向无环图（DAG）。写成数学形式：

回到后厨比喻：节点就是“切菜”“炒菜”这些灶台，边就是“把切好的菜递到炒锅旁”的传送带。只要把这两大要素画清楚，后面所有优化才能展开。

2. 节点身份

流程图上每个工序有不同的职责：备料、切菜、炒菜、装盘；数据流图上的节点也分身份。节点类型是编译器理解“这一步能不能被替换、能不能合并”的重要依据。

最常见的节点是算子节点，它描述计算本身，例如矩阵乘法、卷积、激活函数。其次有常量节点和变量节点：常量保存固定权重，变量保存训练中会更新的参数。还有输入/输出节点，负责和外部交换数据。部分框架还有控制节点，用来表达分支、循环等流程控制。

可以把节点类型看成一个标签函数：

在后厨里，备料、切菜、炒菜的“身份”决定了谁可以提前做、谁必须等。数据流图中的节点身份也决定了编译器能对它们做哪类优化。

3. 边与依赖

传送带不只会运菜，还能告诉工人：必须先放好料，下一步才能开工。数据流图中的边，正是这种“依赖关系”的载体。

边分为两类。数据边携带张量，把上游算子的输出传给下游算子；控制边不携带张量，只约束执行先后顺序，比如“必须先初始化显存，再运行卷积”。有了边，整个图形成偏序关系，编译器才能确定执行计划。

形状传播也沿着边进行：输入张量形状一路往前推断，得到每个中间算子的输出形状。为了让所有依赖都有先来后到，我们需要拓扑排序：

后厨的传送带严格规定了“先切菜，再下锅”。拓扑排序就是给每一位厨师一张排班表，保证每道菜都不会在材料没到齐时提前开工。

4. 构建与校验

主厨拿到菜谱后会先画成工序图，画完还要检查有没有漏掉前序步骤。编译器前端做的是同一件事：把框架里的模型定义翻译成数据流图。

构建过程大致是：前端解析用户代码 → 遇到算子就创建节点 → 把算子间数据传递记录成边 → 补充输入输出节点。TensorFlow 早期通过 Python API 直接构建 GraphDef；PyTorch 的 eager 模式则在运行时由 autograd 逐步记录。无论哪种方式，建图完成后都要做两项校验：拓扑排序查环和静态形状校验。

一个数据流图是否合法，关键看它是不是 DAG。写成逻辑等价关系：

如果后厨流程图中出现了“炒菜要等装盘，装盘又要等炒菜”的死循环，这顿饭永远做不完。拓扑排序就是那个及时喊停的质检员。

5. 优化与代码生成

厨师长拿到流程图后会合并工序、去掉空转等待。AI 编译器基于数据流图做一系列 pass（优化遍），让模型跑得更快、占内存更少。

常见的图优化包括：常量折叠（先算好固定结果）、公共子表达式消除（相同计算只做一次）、算子融合（把卷积、批归一化、激活合并成一个算子）、内存复用（根据张量生命周期复用缓冲区）。每一步优化都会产生一张新的数据流图，但外部看到的输入输出不变。最后经过代码生成，把图里的算子映射到 GPU/NPU 上的实际内核。

优化过程可以写成：

后厨里，把“调味”和“翻炒”合并到一个灶台，既省时间又少洗一个锅。编译器里的算子融合，就是为了少一次读内存、少一次 kernel 启动。

6. 框架实现差异

不同餐厅管理流程图的方式不同。一家要求所有菜必须先写成标准流程图再开火；另一家允许厨师边做菜边记下新步骤。前者像 TensorFlow 的静态图，后者像 PyTorch 的动态图。

TensorFlow 早期默认先把模型构造成静态数据流图，再交给会话执行。静态图便于编译期优化，但调试不够直观。PyTorch 默认采用 eager 模式，每执行一行代码就动态构建一块图，灵活、好调试，但每次迭代都要重复建图，带来额外开销。

为了兼得二者，PyTorch 推出了 TorchScript 和 torch.fx，可以把动态模型“追踪”成静态数据流图；ONNX 则是一种通用图格式，让不同框架之间互转模型。动态图转静态图可看成：

后厨里，动态图是大厨自由的“边想边做”，静态图是标准化门店的“按卡出餐”。TorchScript 就像把大厨的即兴发挥拍成教学视频，之后每个学徒都能照做，还更容易算清楚要买多少食材。

AI 应用场景

Python 演示

class DataFlowGraph:    def __init__(self):        self.nodes = {}        self.nexts = {}        self.prevs = {}     def add_node(self, name, op_type):        self.nodes[name] = op_type        self.nexts[name] = []        self.prevs[name] = []     def add_edge(self, src, dst):        self.nexts[src].append(dst)        self.prevs[dst].append(src)     def topo_sort(self):        indegree = {name: 0 for name in self.nodes}        for src in self.nexts:            for dst in self.nexts[src]:                indegree[dst] += 1        queue = [name for name in self.nodes if indegree[name] == 0]        order = []        while queue:            cur = queue.pop(0)            order.append(cur)            for dst in self.nexts[cur]:                indegree[dst] -= 1                if indegree[dst] == 0:                    queue.append(dst)        if len(order) != len(self.nodes):            raise ValueError("cycle detected")        return order     def run_node(self, name, values):        op = self.nodes[name]        if op == "input" or op == "const":            return values[name]        if op == "mul":            a = self.prevs[name][0]            b = self.prevs[name][1]            return values[a] * values[b]        if op == "add":            a = self.prevs[name][0]            b = self.prevs[name][1]            return values[a] + values[b]     def execute(self, inputs):        values = dict(inputs)        for name in self.topo_sort():            if name not in values:                values[name] = self.run_node(name, values)        return values  def main():    g = DataFlowGraph()    g.add_node("x", "input")    g.add_node("w", "const")    g.add_node("b", "const")    g.add_node("mul", "mul")    g.add_node("add", "add")    g.add_edge("x", "mul")    g.add_edge("w", "mul")    g.add_edge("mul", "add")    g.add_edge("b", "add")    result = g.execute({"x": 3.0, "w": 2.0, "b": 1.0})    print("mul =", result["mul"], "add =", result["add"])  if __name__ == "__main__":    main()

1. g.add_node("x", "input")注册一个输入节点，相当于在流程图上画一个“备料”工序。
2. g.add_edge("x", "mul")建立一条从 \(x\) 到 \(mul\) 的数据边，告诉编译器结果要流向乘法算子。
3. indegree = {name: 0 for name in self.nodes}统计每个节点的前驱数量，前驱越少说明越可能先执行。
4. queue = [name for name in self.nodes if indegree[name] == 0]没有前驱的节点可以放进“待执行队列”。
5. order.append(cur)每处理一个节点，就是为它在流水线上安排好一个位置。
6. for dst in self.nexts[cur]:当前节点完成以后，通知下游：它的前置依赖又少了一个。
7. if len(order) != len(self.nodes):如果最终顺序装不下所有工序，说明流程图里存在环，不能执行。
8. values[name] = self.run_node(name, values)按拓扑序真正执行算子，模拟数据在这张图上流动。

常见误区

面试题