连通性、最小生成树与最大流算法实践

引言

连通性、最小生成树（Minimum Spanning Tree, MST）、最大流（Maximum Flow）在教科书里分属三章，但在图数据库的工程现场它们经常同时出现：供应链里判断两个仓库是否还有可达的调拨路径（连通性）、用最小代价把所有分仓并入同一配送网络（MST）、在带宽或产能受限的情况下算出某个节点最多能往终点推送多少货（最大流），以及反过来找出这条链路真正的瓶颈边（最小割）。这三类问题的共同点是：算法本身都不难，难的是规模、动态性与语义——图上亿条边、边会不断增删、权重是业务口径而非纯数值。

本文按「问题 → 算法 → 复杂度 → 工程取舍 → 落地写法」的顺序把这三类算法过一遍，重点不在推导，而在什么时候选哪个实现、在什么规模下会失效、在 Cypher 与 GDS 里怎么写。前置阅读可参考 图算法实战 与 图路径搜索与最短路算法工程化 ；算法层面的通用复杂度分析可对照 图论与图算法基础 。

1. 连通性判定与并查集

1.1 无向图的连通分量

最朴素的判法是 BFS/DFS 染色：从任一未访问节点出发，能走到的全部染成同一编号，一次遍历 O(n + m)。在 Cypher 里对中等规模图可以直接用变长路径配合 collect(DISTINCT) 近似，但一旦图到千万级，单次全图遍历就会打爆内存。

工程上真正高频的问题不是「有几个连通分量」，而是**「u 和 v 现在是否连通」**，且边在动态增删。这时候静态遍历完全不够用。

1.2 并查集与增量连通

并查集（Union-Find / Disjoint Set Union, DSU）是处理动态连通性的标准结构，核心是路径压缩（Path Compression）与按秩合并（Union by Rank）。两个优化同时用，单次操作的均摊复杂度是反阿克曼函数 α(n)，实际可视为常数：

class DSU:
    def __init__(self, n):
        self.parent = list(range(n))
        self.rank = [0] * n
        self.count = n  # 当前连通分量数

    def find(self, x):
        # 路径压缩：迭代写法，避免深递归爆栈
        root = x
        while self.parent[root] != root:
            root = self.parent[root]
        while self.parent[x] != root:
            self.parent[x], x = root, self.parent[x]
        return root

    def union(self, a, b):
        ra, rb = self.find(a), self.find(b)
        if ra == rb:
            return False
        if self.rank[ra] < self.rank[rb]:
            ra, rb = rb, ra
        self.parent[rb] = ra
        if self.rank[ra] == self.rank[rb]:
            self.rank[ra] += 1
        self.count -= 1
        return True

关键工程点：

• 只支持合并，不支持拆分。删边场景必须换用动态连通性结构（Euler Tour Tree、Link-Cut Tree）或定期重建，重建周期取决于删边比例。
• 不支持带权/带约束的连通（例如「只走金额 ≥ 10 万的边」）。这类条件连通需要按阈值分层维护多个 DSU，或退化为带条件的 BFS。
• 内存是 n 个 int，比邻接表小一到两个数量级，是百万级节点里最省的结构。

1.3 有向图的强连通分量

有向图要区分「弱连通」（忽略方向连通）与「强连通」（双向可达）。强连通分量（Strongly Connected Component, SCC）在风控里极其重要：一个 SCC 内的账户互相转账成环，往往是团伙洗钱的强信号。两个经典算法：

• Kosaraju：两次 DFS，先按完成时间逆序，再在反向图上遍历。思路直观，但要存反向图（内存翻倍）。
• Tarjan：一次 DFS 用栈 + dfn/low 数组，无需反向图。工程上更常用。

def tarjan_scc(n, adj):
    dfn = [0] * n          # 访问时间戳
    low = [0] * n          # 能回溯到的最小时间戳
    on_stack = [False] * n
    stack, sccs = [], []
    timer = [0]

    def dfs(u):
        timer[0] += 1
        dfn[u] = low[u] = timer[0]
        stack.append(u)
        on_stack[u] = True
        for v in adj[u]:
            if dfn[v] == 0:
                dfs(v)
                low[u] = min(low[u], low[v])
            elif on_stack[v]:
                low[u] = min(low[u], dfn[v])
        if low[u] == dfn[u]:        # u 是 SCC 的根
            comp = []
            while True:
                w = stack.pop()
                on_stack[w] = False
                comp.append(w)
                if w == u:
                    break
            sccs.append(comp)

    for i in range(n):
        if dfn[i] == 0:
            dfs(i)
    return sccs

生产实现务必改成迭代版 DFS：递归深度等于最长路径长度，社交图上几十万层递归会直接触发 Python 的 RecursionError 或 C 栈溢出。另外 Tarjan 输出的是逆拓扑序，把每个 SCC 缩成一个点后得到的就是 DAG（缩点图），后续的拓扑排序、最长链、关键路径都能在这个 DAG 上做。

2. 最小生成树

MST 的目标是「用最小总代价把图连通」。注意前提：图必须连通，否则求的是最小生成森林。两条经典路线：

2.1 Kruskal：边排序 + 并查集

按权重升序扫描所有边，若两端不在同一分量就加入：

def kruskal(n, edges):
    # edges: [(w, u, v), ...]
    edges.sort()
    dsu = DSU(n)
    total, picked = 0, []
    for w, u, v in edges:
        if dsu.union(u, v):
            total += w
            picked.append((u, v, w))
            if len(picked) == n - 1:
                break
    return total, picked

复杂度 O(m log m)，瓶颈在排序。优点是天然处理稀疏图，且能直接处理「边不连通」的情况（返回森林）。缺点是每次都要全量排序，动态加边代价高。

2.2 Prim：优先队列 + 已选集合

从一个起点开始，每次把「连接已选集合与未选集合的最小边」拉进来：

import heapq

def prim(n, adj, start=0):
    # adj: {u: [(v, w), ...]}
    visited = [False] * n
    heap = [(0, start, -1)]
    total, picked = 0, []
    while heap:
        w, u, p = heapq.heappop(heap)
        if visited[u]:
            continue
        visited[u] = True
        if p != -1:
            total += w
            picked.append((p, u, w))
        for v, w2 in adj[u]:
            if not visited[v]:
                heapq.heappush(heap, (w2, v, u))
    return total, picked

复杂度 O(m log n)（二叉堆）或 O(m + n log n)（斐波那契堆，实际很少用）。优点是增量友好：往已生成的树上加节点时，只要把新节点的出边压进堆即可。

2.3 工程取舍

选型经验：边数 m < 10⁷ 且内存允许存边表，优先 Kruskal（简单、可并行排序）；图常驻内存且反复做局部扩展，用 Prim。带约束的 MST（例如每条边的权重还要满足某个业务上限）不能直接套这两个算法，需要变成带约束的优化问题。

2.4 几个常见变体

• 最大生成树：把权重取负，或 Kruskal 改成降序扫描。供应链里用它求「最粗壮的骨干网络」，与最小生成树互补。
• 度约束 MST：限制每个节点最多连 k 条边（例如每个分仓最多直连 3 个上游）。这是 NP-hard，工程上要么用贪心近似，要么退化为二分图上的最优匹配。
• 最小生成森林：图不连通时 Kruskal 自然返回森林，注意最后要检查 len(picked) 是否等于 n - c（c 为分量数），否则说明边表本身缺数据。
• 动态 MST：边权重变化时，可以只重算受影响的环——加一条边必然成环，删掉环上最大的边即可维持最小性（环性质）。这条性质是增量维护 MST 的基础。

MST 的一个反直觉之处：它优化的是全网络总代价，而不是任意两点间的最短路。所以用 MST 当路由表会得到「总长最小但单跳绕远」的结果，这两类问题不能混用。

3. 最大流与最小割

最大流问题是：在容量网络里，从源点 s 到汇点 t 最多能推送多少流量。它的对偶问题是最小割，两者满足最大流最小割定理——最大流值等于最小割容量。这条定理是工程上最有用的部分：算最大流的同时顺手拿到了瓶颈边集合。

3.1 Ford-Fulkerson 与 Edmonds-Karp

Ford-Fulkerson 的思路是不断在残量网络里找增广路（Augmenting Path），直到找不到为止。它本身没有规定找路方式，用 DFS 找路时最坏复杂度可以退化到 O(m·f)（f 为最大流值，整数权重下）——在容量是 10⁹ 的图上直接爆炸。

Edmonds-Karp 固定用 BFS 找增广路，复杂度收紧到 O(n·m²)，与容量大小无关。这是能落地的最低门槛实现。

3.2 Dinic：分层图 + 当前弧优化

生产环境首选 Dinic。它先用 BFS 把残量网络分层，再用 DFS 只沿「层数 +1」的边推送，一轮分层可多次增广；每轮至少让最短增广路长度 +1，最多 n 轮：

from collections import deque

class Dinic:
    def __init__(self, n):
        self.n = n
        self.to, self.cap, self.nxt = [], [], []
        self.head = [-1] * n

    def add_edge(self, u, v, c):
        self.to.append(v); self.cap.append(c); self.nxt.append(self.head[u])
        self.head[u] = len(self.to) - 1
        self.to.append(u); self.cap.append(0); self.nxt.append(self.head[v])
        self.head[v] = len(self.to) - 1

    def bfs(self, s, t):
        self.level = [-1] * self.n
        self.level[s] = 0
        q = deque([s])
        while q:
            u = q.popleft()
            e = self.head[u]
            while e != -1:
                if self.cap[e] > 0 and self.level[self.to[e]] < 0:
                    self.level[self.to[e]] = self.level[u] + 1
                    q.append(self.to[e])
                e = self.nxt[e]
        return self.level[t] >= 0

    def dfs(self, u, t, f):
        if u == t:
            return f
        e = self.it[u]
        while e != -1:
            v = self.to[e]
            if self.cap[e] > 0 and self.level[v] == self.level[u] + 1:
                d = self.dfs(v, t, min(f, self.cap[e]))
                if d > 0:
                    self.cap[e] -= d
                    self.cap[e ^ 1] += d
                    return d
            e = self.nxt[e]
            self.it[u] = e
        return 0

    def max_flow(self, s, t):
        flow = 0
        while self.bfs(s, t):
            self.it = self.head[:]     # 当前弧优化：每层只从上次位置继续
            while True:
                f = self.dfs(s, t, float('inf'))
                if f == 0:
                    break
                flow += f
        return flow

关键优化点：

• 反向边用 e ^ 1 成对存放，省掉显式反向边索引，也方便回退流量。
• 当前弧优化（Current Arc）：it[u] 记录该节点上次扫描到的边，避免重复扫描已饱和的边，是常数级优化的主要来源。
• 多路增广：一次 DFS 尽量把流量推满再返回，而不是找到一条路就返回。
• 单位容量网络（所有边容量为 1）上 Dinic 复杂度是 O(m√n)，做二分图匹配时比匈牙利算法更快。

3.3 常见建模技巧

真实问题往往不是标准网络流，需要转换：

点容量拆点是最常被忽略的一步。比如「每个中转仓每天最多处理 500 单」，如果直接把容量挂到边上，模型就错了。

下面是一个典型的「任务分配」建模（二分图匹配 + 容量约束），把 N 个订单分给 M 个骑手，每个骑手最多接 cap 单：

def build_assignment(order_ids, rider_ids, edges, rider_cap):
    # 节点编号：0 = 超级源，1..N = 订单，N+1..N+M = 骑手，N+M+1 = 超级汇
    n = 1 + len(order_ids) + len(rider_ids) + 1
    src, sink = 0, n - 1
    d = Dinic(n)
    for i in range(len(order_ids)):
        d.add_edge(src, 1 + i, 1)             # 每个订单只能被满足一次
    for (o, r) in edges:
        d.add_edge(1 + o, 1 + len(order_ids) + r, 1)
    for j in range(len(rider_ids)):
        d.add_edge(1 + len(order_ids) + j, sink, rider_cap[j])
    return d, src, sink

最大流值就是「最多能完成的订单数」，而残量网络里的饱和边直接给出了具体分配方案（哪条 order→rider 边流量为 1）。这种「求最大匹配同时拿到方案」的能力，是网络流比单纯二分图匹配更通用的地方。

3.4 费用流与带权场景

当目标不是「流最大」而是「在最大流前提下费用最小」（例如订单分配还要最小化总配送距离），就要用最小费用最大流（Min-Cost Max-Flow, MCMF）。做法是把 Dinic 的分层 BFS 换成最短路（SPFA 或 Johnson 势能法 + Dijkstra），每次沿最短路增广：

MCMF 要点：
1. 每条边带 (cap, cost)，反向边 cost 取负
2. 反复找「单位费用最小的增广路」→ SPFA / 势能 Dijkstra
3. 沿该路推尽量多的流量，累加 cost
4. 直到不存在增广路
陷阱：负费用边导致 Dijkstra 失效 → 必须用势能（potential）预处理
复杂度：O(F · m log n)，F 为最大流值；F 大时很慢

MCMF 在运筹类问题里很常见（车辆调度、人员排班、广告投放），但它是伪多项式算法，流量值大时会退化，需要控制规模或改用线性规划。

4. 割与瓶颈分析

4.1 最小割定位瓶颈

跑完最大流后，在残量网络里从源点做一次 BFS/DFS，能到达的节点集合记为 S，其余记为 T，则割集就是所有从 S 指向 T 的原始边。这些边就是真正的瓶颈：想提升整体吞吐，扩容对象只可能是它们。

工程价值在于：它把「哪条边该扩容」从拍脑袋变成了可证明的结论。对供应链、带宽规划、任务调度都直接可用。

4.2 全局最小割：Stoer-Wagner

最大流求的是指定 s-t 之间的最小割。如果想知道「整张图最脆弱的一刀在哪」（不指定源汇），要用 Stoer-Wagner 算法，复杂度 O(n³) 或 O(nm log n)：

Stoer-Wagner 核心：
1. 维护已合并集合 A，初始为空
2. 反复把「到 A 的总权重最大」的顶点加入 A（类似 Prim 的最大生成树变体）
3. 最后加入的两个顶点 s, t，记录当前割值 cut = 所有连到 t 的权重和
4. 用 cut 更新全局最小值，然后把 s, t 合并成一个点
5. 重复直到只剩一个点

适用场景：网络可靠性分析（找最容易把网络劈成两半的边集）、社区划分的辅助判据、聚类质量评估。注意它是 O(n³)，n 上万就基本跑不动，需要采样或先做图缩减。

5. 在 Cypher 与 GDS 中落地

5.1 用 Cypher 表达连通性

小图上的条件连通（带属性过滤）可以直接用变长路径：

// 判断两个仓库之间是否存在运力充足的调拨路径（最多 6 跳）
MATCH (a:Warehouse {code: 'WH-A'}), (b:Warehouse {code: 'WH-B'})
MATCH p = (a)-[:SHIP*1..6]->(b)
WHERE ALL(r IN relationships(p) WHERE r.capacity >= 1000)
RETURN count(p) > 0 AS reachable
LIMIT 1

注意 LIMIT 1 配合 count(p) > 0 在 Neo4j 里不会提前终止（聚合要先收集），真要短路应改用 EXISTS { ... } 子查询或 CALL { ... } 里的 LIMIT 1。

5.2 GDS 的图投影与算法调用

GDS（Graph Data Science）把算法跑在内存投影图上，避免直接扫存储：

// 1. 建立内存投影
CALL gds.graph.project(
  'supply',
  ['Warehouse', 'Hub'],
  {SHIP: {type: 'SHIP', properties: 'capacity'}},
  {undirectedRelationshipTypes: ['SHIP']}
);

// 2. 弱连通分量（WCC）
CALL gds.wcc.stream('supply')
YIELD nodeId, componentId
RETURN gds.util.asNode(nodeId).code AS code, componentId
ORDER BY componentId;

// 3. 强连通分量（SCC，需要 directed 投影）
CALL gds.graph.project('supply_d', ['Warehouse'], {SHIP: {type: 'SHIP', orientation: 'NATURAL'}});
CALL gds.scc.stream('supply_d') YIELD nodeId, componentId RETURN nodeId, componentId;

// 4. 用 Kruskal 求最小生成树（GDS 提供 k-spanning-tree）
CALL gds.spanningTree.stream('supply', {relationshipWeightProperty: 'capacity'})
YIELD nodeId, parentId, weight
RETURN gds.util.asNode(nodeId).code AS node, weight;

// 5. 用完释放投影，否则内存不会回收
CALL gds.graph.drop('supply');

5.3 最大流在 GDS 中的现状

GDS 内置算法集里没有通用最大流。生产上通常有两条路：一是把图导出成边表，在外部用 NetworkX（nx.maximum_flow）或 OR-Tools 求解，再把结果写回；二是把问题建模成线性规划交给专门的求解器。选外部求解时要留意导出/写回的开销——千万级边表序列化本身就是分钟级操作。

导出并回写结果的最小闭环：

import neo4j, networkx as nx

driver = neo4j.GraphDatabase.driver(URI, auth=(USER, PWD))
with driver.session() as s:
    edges = s.run("""
        MATCH (a:Warehouse)-[r:SHIP]->(b:Warehouse)
        RETURN a.code AS u, b.code AS v, r.capacity AS c
    """).data()

G = nx.DiGraph()
for e in edges:
    G.add_edge(e['u'], e['v'], capacity=e['c'])

value, flow = nx.maximum_flow(G, 'WH-SRC', 'WH-SINK', capacity='capacity')

# 把流量写回边属性，便于在图上做瓶颈可视化
with driver.session() as s:
    for u, outs in flow.items():
        for v, f in outs.items():
            if f > 0:
                s.run("""
                    MATCH (a:Warehouse {code:$u})-[r:SHIP]->(b:Warehouse {code:$v})
                    SET r.flow = $f
                """, u=u, v=v, f=f)

回写时用 UNWIND 批量提交比逐条 SET 快一到两个数量级，几万条边应改成 UNWIND $rows AS row MATCH ... SET r.flow = row.f。算法输出的连通分量/割集同理，最好写回成节点属性或独立的关系类型，方便后续在可视化工具里直接着色，而不是每次重算。相关批量写入手法可参考 图导入与 ETL 。

6. 排错与调优

现象一：并查集结果与全图遍历不一致。 九成是漏了「孤立节点」。并查集初始化 count = n 时把从未出现在边表里的节点也算了进去，若业务口径是「只统计有边的节点」，需要在建 DSU 时只登记出现过的节点。

现象二：Tarjan 栈溢出。 递归深度 = 最长路径长度。改成迭代版，或先对图做一次缩点降低深度。

现象三：最大流跑得慢。 先确认不是 Edmonds-Karp 用在容量极大的图上（那是容量相关的伪多项式）。再看是否漏了当前弧优化。最后检查是否用了浮点容量——浮点会导致残量出现 1e-12 级别的残留，算法永远找不到终止条件，容量一律用整数。

现象四：最小割结果里出现了容量为 0 的边。 说明反向边被当成了原图边。定位割集时要按原始边的方向过滤，只取 S→T 的原始边，忽略残量反向边。

小结

这三类算法在工程上的分水岭是规模与动态性：百万级以内的静态图，Kruskal + Tarjan + Dinic 这套组合足够；边频繁增删时，连通性必须换成并查集或 Link-Cut Tree；图大到内存装不下时，最大流这类需要全局状态的问题就得外置求解器。选型前先回答三个问题——图多大、边动不动、容量是不是整数——答案基本就唯一了。落地时优先用 GDS 的内存投影跑内置算法，只有在 GDS 没有对应实现（如最大流）时才导出到外部求解，并把导出开销计入预算。