3、Python实现DAG：使用NetworkX构建DAG、节点与边的增删改查、可视化DAG

理论讲完了，咱们得动真格的了。这一章，我会带你用 Python 把 DAG 实实在在地跑起来。

你可能会问：为什么要用 NetworkX？说实话，Python 里能处理图的库不少，但 NetworkX 是最「亲民」的一个。它内置了 DAG 的专用方法，比如拓扑排序、有向环检测，这些在任务编排里都是高频操作。我个人习惯把它当作 DAG 的「瑞士军刀」——轻量、够用、社区活跃。

3.1 安装与初始化：三行代码搞定

先装包。没什么好说的，一行命令：

pip install networkx matplotlib

然后，在 Python 里创建一个 DAG 对象：

import networkx as nx

# 创建一个有向图
dag = nx.DiGraph()

# 明确声明这是一个 DAG（其实不声明也行，但写上更严谨）
# 注意：DiGraph 本身不强制无环，需要我们自己保证
print(type(dag))  # <class 'networkx.classes.digraph.DiGraph'>

嗯，这里要注意：DiGraph 只是一个有向图容器，它不会自动阻止你加环。DAG 的「无环」约束，需要我们在增删改查时自己维护。后面我会讲怎么用 NetworkX 检测环。

3.2 节点与边的增删改查

这部分是日常操作的核心。我把它拆成四个动作来讲：增、删、改、查。

3.2.1 增加节点与边

加节点很简单：

# 添加单个节点
dag.add_node("task_a")

# 添加多个节点
dag.add_nodes_from(["task_b", "task_c", "task_d"])

# 添加边（同时会自动创建不存在的节点）
dag.add_edge("task_a", "task_b")  # task_a -> task_b
dag.add_edges_from([
    ("task_b", "task_c"),
    ("task_c", "task_d"),
    ("task_a", "task_d")  # 跨层依赖，没问题
])

你看，add_edge 如果引用了不存在的节点，NetworkX 会自动帮你创建。这个特性很方便，但也容易埋坑——我曾经因为拼写错误（比如 task_a 写成了 task_A），导致图里凭空多出一个孤立节点，排查了半天。

3.2.2 删除节点与边

删除操作同样直观：

# 删除单个节点（关联的边也会被自动删除）
dag.remove_node("task_d")

# 删除多个节点
dag.remove_nodes_from(["task_c", "task_b"])

# 删除单条边
dag.remove_edge("task_a", "task_b")

# 删除多条边
dag.remove_edges_from([("task_a", "task_b"), ("task_b", "task_c")])

这里有个细节：remove_node 会级联删除所有关联的边。如果你只想断开连接但保留节点，应该用 remove_edge。我在做任务编排时，经常需要「保留节点但调整依赖关系」，这时候 remove_edge 就派上用场了。

3.2.3 修改节点与边

NetworkX 的「改」操作，本质上是通过「删 + 增」实现的。比如你想修改一个节点的名字：

# 修改节点名：先记录邻居，再删旧节点，再加新节点
neighbors = list(dag.predecessors("task_a")) + list(dag.successors("task_a"))
dag.remove_node("task_a")
dag.add_node("task_a_new")
for pred in dag.predecessors("task_a"):  # 这里其实已经空了，因为节点已删
    pass
# 正确做法：重新建立连接
for pred in neighbors:
    if pred in dag.nodes():
        dag.add_edge(pred, "task_a_new")
    else:
        dag.add_edge("task_a_new", pred)  # 根据方向判断

说实话，这个操作有点繁琐。我个人建议：如果只是改名字，不如直接删了重建。或者，你可以给节点加属性来「标记」而不改名字：

# 给节点加属性
dag.nodes["task_a"]["status"] = "completed"
dag.nodes["task_a"]["priority"] = 1

# 给边加属性
dag.edges["task_a", "task_b"]["weight"] = 0.5

这样既保留了图结构，又能灵活调整元数据。我在实际项目中，经常用节点属性来存储任务的执行状态、重试次数、超时时间等。

3.2.4 查询节点与边

查询是日常使用最频繁的操作。NetworkX 提供了丰富的查询接口：

# 查询所有节点
print(dag.nodes())  # ['task_a', 'task_b', ...]

# 查询所有边
print(dag.edges())  # [('task_a', 'task_b'), ...]

# 查询某个节点的前驱（依赖谁）
print(list(dag.predecessors("task_c")))  # ['task_b']

# 查询某个节点的后继（谁依赖我）
print(list(dag.successors("task_c")))  # ['task_d']

# 查询入度（有多少个直接前驱）
print(dag.in_degree("task_c"))  # 1

# 查询出度（有多少个直接后继）
print(dag.out_degree("task_c"))  # 1

# 判断是否存在某条边
print(dag.has_edge("task_a", "task_b"))  # True

# 判断是否存在某个节点
print(dag.has_node("task_e"))  # False

这些接口在任务调度时特别有用。比如，你想找出所有「没有前驱」的任务（即入度为 0 的节点），这些就是可以立即执行的任务：

ready_tasks = [node for node in dag.nodes() if dag.in_degree(node) == 0]
print(ready_tasks)

3.3 环检测：DAG 的生命线

前面说过，DiGraph 不阻止加环。但 DAG 的核心就是「无环」，所以环检测是必须的。

# 检测是否有环
print(nx.is_directed_acyclic_graph(dag))  # True 表示无环

# 如果存在环，找出环的路径
try:
    cycle = nx.find_cycle(dag)
    print("发现环:", cycle)
except nx.NetworkXNoCycle:
    print("无环，安全")

我曾经在一个生产环境里，因为配置文件的依赖写反了（A 依赖 B，B 又依赖 A），导致调度器死循环。从那以后，我每次修改 DAG 后都会强制跑一次环检测，写成 CI 流程的一部分。

3.4 可视化 DAG：一图胜千言

代码里的 DAG 再正确，也不如一张图直观。NetworkX 本身不带可视化引擎，但它可以无缝对接 Matplotlib。

import matplotlib.pyplot as plt

# 绘制 DAG
pos = nx.spring_layout(dag, seed=42)  # 布局算法，seed 保证每次位置一致
nx.draw(dag, pos, with_labels=True, 
        node_color='lightblue', 
        edge_color='gray',
        node_size=2000,
        font_size=12,
        arrows=True,
        arrowsize=20)

plt.title("My First DAG")
plt.show()

这段代码会弹出一个窗口，显示节点和边的图形。不过，默认的 spring_layout 对于 DAG 来说，布局可能不够「层次化」。我更喜欢用 hierarchical_pos 或者 dot 布局，让依赖关系从上到下清晰展示。

下面是我手绘的一张 DAG 结构图，展示了节点、边、前驱/后继的关系：

你看，这张图里 task_a 没有前驱（入度为 0），所以它可以最先执行。task_f 没有后继（出度为 0），是终点节点。这种「从入度为 0 的节点开始，逐层推进」的思路，就是任务调度的核心逻辑。

3.5 实战：构建一个简单的任务 DAG

最后，咱们把上面的知识点串起来，写一个完整的例子：

import networkx as nx
import matplotlib.pyplot as plt

# 1. 创建 DAG
dag = nx.DiGraph()

# 2. 添加任务节点
tasks = ["数据采集", "数据清洗", "特征工程", "模型训练", "模型评估", "部署上线"]
dag.add_nodes_from(tasks)

# 3. 添加依赖边
dependencies = [
    ("数据采集", "数据清洗"),
    ("数据清洗", "特征工程"),
    ("特征工程", "模型训练"),
    ("模型训练", "模型评估"),
    ("模型评估", "部署上线"),
    ("数据采集", "特征工程"),  # 跨层依赖
]
dag.add_edges_from(dependencies)

# 4. 环检测
if not nx.is_directed_acyclic_graph(dag):
    print("错误：存在环！")
    exit(1)

# 5. 拓扑排序
topo_order = list(nx.topological_sort(dag))
print("执行顺序:", topo_order)

# 6. 可视化
pos = nx.spring_layout(dag, seed=42)
nx.draw(dag, pos, with_labels=True, 
        node_color='lightgreen',
        edge_color='gray',
        node_size=3000,
        font_size=10,
        arrows=True,
        arrowsize=20)
plt.title("任务编排 DAG")
plt.show()

这段代码跑完之后，你会看到一个清晰的任务依赖图。拓扑排序的结果就是任务的执行顺序——当然，实际调度时还要考虑并行度、资源限制等因素，那是后面章节的内容。