基于优先队列的Prim最小生成树实现错误排查

我用优先队列实现的Prim最小生成树算法，点间距离计算是正确的，但运行以下测试用例时，嵌套循环里的部分值被错误更新为更小值，导致最终结果偏差——预期结果为7.064495102，实际得到7.650281540，怀疑是嵌套循环的实现逻辑有问题。

测试用例 #

5
0 0
0 2
1 1
3 0
3 2

算法实现代码 #

def minimum_distance(x, y):
    result = 0.
    distance = {}
    for i in range(len(x)):
        j=i+1
        while j<len(x):
            distance[i, j] = calc_distance(x[i], y[i], x[j],y[j])
            # distance.append([i, j, calc_distance(x[i], y[i], x[j],y[j])])
            j+=1
            
    cost = []
    parent = []
    for k in range(len(x)):
        cost.append([float('inf'), k])
        parent.append([None, k]) 
    
    # [cost, vertex]
    cost[0] = [0,0]
    parent[0] = [-1,0]

    pq = min_heap(cost)
    cost_pq = pq.heapify()
    
    while len(cost_pq) != 0:
        v, cost_pq = pq.extract_min(cost_pq)
        min_cost, min_current_node = v[0], v[1]
        result += min_cost
        for edge in distance:
            for vertex in cost_pq:
                # check if cost_pq contains vertex edge[1]
                if vertex[1] == edge[1]:
                    vertex_index = cost_pq.index(vertex)
                    if cost_pq[vertex_index][0] > distance[edge[0], edge[1]]:
                        cost_pq[vertex_index][0] = distance[edge[0], edge[1]]
                        parent[edge[1]][0] = edge[0]
                        pq.heapify() 
            
    return result

问题分析 #

你的嵌套循环逻辑存在核心错误，直接导致了结果偏差：

• 遍历范围错误：Prim算法中，每次取出当前代价最小的节点后，只需要遍历该节点的所有邻接边，去更新邻接节点的代价即可。但你现在是遍历所有边+堆中所有顶点，不仅效率极低，还会错误地更新无关节点的代价。
• 边存储不完整：distance字典只存储了i<j的单向边，没有反向的(j,i)条目。当取出的节点是j时，无法找到对应的边来更新i的代价，同时也会导致部分邻接关系被遗漏。
• 堆维护错误：直接修改cost_pq中的元素后调用heapify()是不合理的——heapify()是对整个数组重新建堆，而正确的优先队列在更新元素代价后，应该执行上浮/下沉操作来维护堆结构。另外，cost_pq的修改是否同步到了堆的内部状态也存疑。
• 节点存在性判断低效：用vertex[1] == edge[1]再调用index()查找节点的方式，时间复杂度很高，而且容易出现重复匹配的问题。

修复思路 #

1. 重构邻接表：放弃当前的distance字典，改用邻接表存储每个节点的所有邻接关系，确保双向边都被记录：

   adj = [[] for _ in range(len(x))]
   for i in range(len(x)):
       for j in range(i+1, len(x)):
           dist = calc_distance(x[i], y[i], x[j], y[j])
           adj[i].append((j, dist))
           adj[j].append((i, dist))
   

2. 修正循环逻辑：取出当前节点后，只遍历它的邻接边，针对每个邻接节点判断是否需要更新代价：

   while len(cost_pq) != 0:
       v, cost_pq = pq.extract_min(cost_pq)
       min_cost, u = v[0], v[1]
       result += min_cost
       # 遍历u的所有邻接边
       for (v_nei, dist) in adj[u]:
           # 查找邻接节点在堆中的位置
           for idx, vertex in enumerate(cost_pq):
               if vertex[1] == v_nei:
                   if vertex[0] > dist:
                       cost_pq[idx][0] = dist
                       parent[v_nei][0] = u
                       # 这里需要优先队列支持decrease-key操作，或者重新堆化
                       pq.heapify()
                   break
   

3. 优化堆操作：如果自己实现的min_heap不支持decrease-key，可以改用允许堆中存在重复节点的方式——每次更新代价时直接向堆中添加新的条目，取出节点时判断该节点是否已经被加入生成树（用一个visited数组标记），如果已访问则跳过该条目。

内容的提问来源于stack exchange，提问作者Young42