函数说明
- 算法名称：dijkstra_shortest_path
- 目标：使用优先队列实现 Dijkstra 最短路径算法，支持有向/无向加权图（权重非负）。可计算：
- 起点到所有顶点的最短距离
- 起点到指定终点的最短路并重建路径
- 输入：
- graph: dict[节点, list[(邻接节点, 权重: float)]]
- start: 节点标识（字符串或整数，需可哈希）
- end: 可选节点标识，None 表示计算到所有节点
- return_path: bool，true 时返回路径（当 end 为 None 时返回前驱表用于路径恢复）
- 输出：
- 若 end 为 None：
- return_path = False：返回 dict[节点 -> 最短距离]
- return_path = True：返回 (dict[节点 -> 最短距离], dict[节点 -> 前驱节点])
- 若提供 end：
- return_path = False：返回 distance: float（不可达为 math.inf）
- return_path = True：返回 (distance: float, path: list[节点])（不可达为 (math.inf, [])）
- 约定：
- 支持有向/无向图：无向图请在输入中为每条边提供双向邻接即可
- 权重必须为非负且有限数值
- 空图、孤立顶点可健壮处理
代码实现（可直接运行与单元测试）
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
import heapq
import math
import unittest
def dijkstra_shortest_path(
graph: Dict[Any, Iterable[Tuple[Any, Union[int, float]]]],
start: Any,
end: Optional[Any] = None,
return_path: bool = False,
):
"""
使用优先队列的 Dijkstra 最短路径算法。
参数:
graph: 字典，key 为节点（可哈希），value 为邻接列表 (邻居, 权重)。
示例: {"A":[("B",1.5),("C",2.0)], "B":[("C",1.0)]}
注意：无向图请提供双向边。
start: 起点（可哈希）
end: 终点（可选，None 表示计算到所有节点）
return_path: 若为 True：
- 当 end is None：返回 (distances, predecessors)
- 当 end is not None：返回 (distance, path)
返回:
- 如果 end is None:
- return_path=False -> dict[node] = 最短距离
- return_path=True -> (dict[node] = 最短距离, dict[node] = 前驱节点)
- 如果 end is not None:
- return_path=False -> distance: float（不可达为 math.inf）
- return_path=True -> (distance: float, path: list[节点]；不可达为 (math.inf, []))
健壮性:
- 空图处理: 若图为空，仅返回与 start 相关的结果（到自身 0，其余不存在）
- 孤立顶点: 允许 start 不在图中，视为孤立点；若 end==start，距离为 0，否则不可达
- 校验:
- 节点需可哈希
- 权重需为非负且有限的数值
时间复杂度:
- O(E log V)，其中 V 为顶点数，E 为边数（使用最小堆）
空间复杂度:
- O(V + E) 用于距离、前驱、邻接表和堆
"""
# ------------------------
# 输入校验与规范化
# ------------------------
if graph is None:
raise TypeError("graph 不能为 None，请提供字典形式的邻接表。")
if not isinstance(graph, dict):
raise TypeError("graph 必须为 dict[节点 -> 邻接列表]。")
# 校验并收集节点集；邻接表标准化为 dict[node] -> List[(neighbor, weight)]
adjacency: Dict[Any, List[Tuple[Any, float]]] = {}
nodes = set()
# 校验 start/end 可哈希
try:
hash(start)
except Exception as e:
raise TypeError(f"start 节点不可哈希: {start!r}") from e
if end is not None:
try:
hash(end)
except Exception as e:
raise TypeError(f"end 节点不可哈希: {end!r}") from e
for u, edges in graph.items():
# 校验节点可哈希
try:
hash(u)
except Exception as e:
raise TypeError(f"节点不可哈希: {u!r}") from e
nodes.add(u)
# 支持空/None 邻接列表（视为空）
if edges is None:
edges = []
# 校验邻接列表类型
if not isinstance(edges, (list, tuple)):
raise TypeError(f"节点 {u!r} 的邻接列表必须为 list/tuple[(邻居, 权重)]。")
normalized_edges: List[Tuple[Any, float]] = []
for item in edges:
if not isinstance(item, (list, tuple)) or len(item) != 2:
raise TypeError(f"邻接项需为 (neighbor, weight)，收到: {item!r}")
v, w = item
try:
hash(v)
except Exception as e:
raise TypeError(f"邻居节点不可哈希: {v!r}") from e
# 权重校验：数值、有限、非负
if not isinstance(w, (int, float)):
raise TypeError(f"权重必须为数值 (int/float)，收到: {w!r}")
if not math.isfinite(w):
raise ValueError(f"权重必须为有限数值，收到: {w!r}")
if w < 0:
raise ValueError(f"权重必须非负，收到: {w!r}")
nodes.add(v)
normalized_edges.append((v, float(w)))
adjacency[u] = normalized_edges
# 将 start/end 纳入节点集合（允许孤立点）
nodes.add(start)
if end is not None:
nodes.add(end)
# 确保每个节点在 adjacency 中有条目
for n in nodes:
adjacency.setdefault(n, [])
# ------------------------
# Dijkstra 主过程
# ------------------------
# 初始化距离与前驱
dist: Dict[Any, float] = {n: math.inf for n in nodes}
dist[start] = 0.0
prev: Dict[Any, Any] = {} # 记录最优前驱
# 小顶堆：(距离, 节点)
heap: List[Tuple[float, Any]] = [(0.0, start)]
# 若 end 指定，可提前终止：当 end 首次弹出且距离最小，即已确定最短路
target = end
while heap:
d_u, u = heapq.heappop(heap)
if d_u > dist[u]:
# 堆中旧条目，忽略
continue
if target is not None and u == target:
# 最短路径已确定，提前结束
break
# 松弛
for v, w in adjacency[u]:
nd = d_u + w
if nd < dist[v]:
dist[v] = nd
prev[v] = u
heapq.heappush(heap, (nd, v))
# ------------------------
# 构造返回值
# ------------------------
def reconstruct_path(predecessor: Dict[Any, Any], s: Any, t: Any) -> List[Any]:
if s == t:
return [s]
if dist.get(t, math.inf) == math.inf:
return []
path = []
cur = t
while cur != s:
path.append(cur)
cur = predecessor.get(cur)
if cur is None:
# 理论上不会发生；保护性处理
return []
path.append(s)
path.reverse()
return path
if end is None:
# 需求：返回到所有节点的最短距离；可选返回前驱表
return (dist, prev) if return_path else dist
else:
# 需求：返回到 end 的距离；可选返回路径
distance = dist.get(end, math.inf)
if return_path:
path = reconstruct_path(prev, start, end)
return (distance, path)
else:
return distance
# ------------------------
# 示例调用
# ------------------------
def _demo():
example_graph = {
"A": [("B", 1.5), ("C", 2.0)],
"B": [("C", 1.0)],
# 无向图示例：请提供双向边
# "C": [("A", 2.0), ("B", 1.0)]
}
print("示例：起点到所有节点的距离")
dist_all = dijkstra_shortest_path(example_graph, start="A", end=None, return_path=False)
print(dist_all) # {'A': 0.0, 'B': 1.5, 'C': 2.5}
print("示例：起点到终点的最短路与路径")
distance, path = dijkstra_shortest_path(example_graph, start="A", end="C", return_path=True)
print(distance, path) # 2.5 ['A', 'B', 'C']
# ------------------------
# 单元测试
# ------------------------
class TestDijkstraShortestPath(unittest.TestCase):
def test_directed_basic(self):
g = {
"A": [("B", 1.5), ("C", 2.0)],
"B": [("C", 1.0)],
}
distance, path = dijkstra_shortest_path(g, "A", "C", return_path=True)
self.assertEqual(distance, 2.5)
self.assertEqual(path, ["A", "B", "C"])
def test_all_distances(self):
g = {
"A": [("B", 2), ("C", 5)],
"B": [("C", 1)],
"C": [],
"D": [] # 孤立节点
}
dist = dijkstra_shortest_path(g, "A", end=None, return_path=False)
self.assertEqual(dist["A"], 0.0)
self.assertEqual(dist["B"], 2.0)
self.assertEqual(dist["C"], 3.0)
self.assertTrue(math.isinf(dist["D"]))
dist2, prev = dijkstra_shortest_path(g, "A", end=None, return_path=True)
self.assertEqual(dist2["C"], 3.0)
# 路径恢复 A->B->C
path_c = []
cur = "C"
while cur in prev:
path_c.append(cur)
cur = prev[cur]
path_c.append("A")
path_c.reverse()
self.assertEqual(path_c, ["A", "B", "C"])
def test_unreachable(self):
g = {
1: [(2, 1.0)],
2: [],
3: [] # 不可达
}
distance = dijkstra_shortest_path(g, 1, end=3, return_path=False)
self.assertTrue(math.isinf(distance))
distance, path = dijkstra_shortest_path(g, 1, end=3, return_path=True)
self.assertTrue(math.isinf(distance))
self.assertEqual(path, [])
def test_empty_graph(self):
g = {}
# 起点到自身为 0，其他不存在
dist = dijkstra_shortest_path(g, "S", end=None, return_path=False)
self.assertEqual(dist["S"], 0.0)
# 指定终点且不可达
d, p = dijkstra_shortest_path(g, "S", end="T", return_path=True)
self.assertTrue(math.isinf(d))
self.assertEqual(p, [])
def test_isolated_start(self):
g = {"A": [("B", 2.0)], "B": []}
# start 不在图中，视为孤立点
d = dijkstra_shortest_path(g, "Z", end=None, return_path=False)
self.assertEqual(d["Z"], 0.0)
self.assertTrue(math.isinf(d["A"]))
self.assertTrue(math.isinf(d["B"]))
# end==start
distance, path = dijkstra_shortest_path(g, "Z", end="Z", return_path=True)
self.assertEqual(distance, 0.0)
self.assertEqual(path, ["Z"])
def test_invalid_weight(self):
g_neg = {"A": [("B", -1.0)]}
with self.assertRaises(ValueError):
dijkstra_shortest_path(g_neg, "A", end=None)
g_inf = {"A": [("B", float("inf"))]}
with self.assertRaises(ValueError):
dijkstra_shortest_path(g_inf, "A", end=None)
def test_non_hashable_node(self):
g_bad = {["A"]: [("B", 1.0)]} # 列表不可哈希
with self.assertRaises(TypeError):
dijkstra_shortest_path(g_bad, "A", end=None)
if __name__ == "__main__":
_demo()
unittest.main(argv=["-v"], exit=False)
简要实现思路与逻辑说明
- 使用最小堆（heapq）维护当前已知的最短距离候选，弹出距离最小的节点进行松弛，复杂度 O(E log V)。
- 使用 dist 记录起点到各节点的最短距离；使用 prev 记录到达每个节点的最佳前驱，用于路径重建。
- 当指定了 end 时，首次从堆中弹出 end 即确定其最短路径，可提前结束。
- 输入校验涵盖：
- graph 类型与邻接项格式
- 节点可哈希性与权重非负、有限
- 空图与孤立顶点处理（将 start/end 纳入节点集合，保证算法健壮性）
- 无向图支持通过在输入中提供双向边实现；函数不自动补全双向边以避免误判图类型。
