本文目录导读：
我来详细讲解如何使用NetworkX实现A算法，虽然NetworkX内置了A算法,但我们可以自己实现来理解原理。
基础环境准备
import networkx as nx import matplotlib.pyplot as plt import numpy as np import heapq # 用于优先队列 # 创建图 G = nx.Graph()
创建网格地图示例
# 创建一个5x5的网格地图
def create_grid_graph(rows=5, cols=5, obstacles=None):
G = nx.Graph()
# 添加节点
for i in range(rows):
for j in range(cols):
if obstacles and (i, j) in obstacles:
continue
G.add_node((i, j), pos=(j, -i))
# 添加边（上下左右四个方向）
for i in range(rows):
for j in range(cols):
if obstacles and (i, j) in obstacles:
continue
# 右连接
if j + 1 < cols and (not obstacles or (i, j+1) not in obstacles):
G.add_edge((i, j), (i, j+1), weight=1)
# 下连接
if i + 1 < rows and (not obstacles or (i+1, j) not in obstacles):
G.add_edge((i, j), (i+1, j), weight=1)
return G
# 创建带障碍物的地图
obstacles = [(1, 2), (2, 2), (3, 2), (1, 4)]
G = create_grid_graph(5, 5, obstacles)
# 可视化
pos = {(i, j): (j, -i) for i in range(5) for j in range(5)}
nx.draw(G, pos, with_labels=True, node_color='lightblue',
node_size=500, font_size=8)"网格地图")
plt.show()
手动实现A*算法
def astar_algorithm(G, start, goal, heuristic, weight='weight'):
"""
手动实现A*算法
"""
# 初始化
open_set = [] # 优先队列，存储(f_score, node, path)
heapq.heappush(open_set, (0, start, [start]))
# 记录已访问节点及其g_score
g_scores = {start: 0}
visited = set()
while open_set:
# 取出f_score最小的节点
f_score, current, path = heapq.heappop(open_set)
# 如果到达目标
if current == goal:
return path, g_scores[current]
# 如果已经访问过，跳过
if current in visited:
continue
visited.add(current)
# 遍历邻居节点
for neighbor in G.neighbors(current):
if neighbor in visited:
continue
# 计算新的g_score
edge_weight = G[current][neighbor].get(weight, 1)
new_g_score = g_scores[current] + edge_weight
# 如果找到更好的路径
if neighbor not in g_scores or new_g_score < g_scores[neighbor]:
g_scores[neighbor] = new_g_score
f_score = new_g_score + heuristic(neighbor, goal)
new_path = path + [neighbor]
heapq.heappush(open_set, (f_score, neighbor, new_path))
return None, float('inf')
# 启发式函数：曼哈顿距离
def manhattan_distance(a, b):
return abs(a[0] - b[0]) + abs(a[1] - b[1])
# 欧几里得距离
def euclidean_distance(a, b):
return ((a[0] - b[0])**2 + (a[1] - b[1])**2)**0.5
使用NetworkX内置的A*算法
# NetworkX内置的astar_path
def networkx_astar(G, start, goal, heuristic=None):
if heuristic is None:
heuristic = manhattan_distance
path = nx.astar_path(G, start, goal, heuristic=heuristic)
cost = nx.astar_path_length(G, start, goal, heuristic=heuristic)
return path, cost
# 使用示例
start = (0, 0)
goal = (4, 4)
# 使用自定义实现
path1, cost1 = astar_algorithm(G, start, goal, manhattan_distance)
print(f"自定义A*路径: {path1}")
print(f"自定义A*成本: {cost1}")
# 使用NetworkX内置函数
path2, cost2 = networkx_astar(G, start, goal, manhattan_distance)
print(f"NetworkX A*路径: {path2}")
print(f"NetworkX A*成本: {cost2}")
可视化路径
def visualize_path(G, path, title="A* Path Finding"):
# 获取节点位置
pos = {(i, j): (j, -i) for i in range(5) for j in range(5)}
plt.figure(figsize=(10, 8))
# 绘制所有节点和边
nx.draw(G, pos, with_labels=True, node_color='lightblue',
node_size=500, font_size=8, edge_color='gray')
if path:
# 绘制路径
path_edges = list(zip(path[:-1], path[1:]))
nx.draw_networkx_nodes(G, pos, nodelist=[path[0]],
node_color='green', node_size=600, label='起点')
nx.draw_networkx_nodes(G, pos, nodelist=[path[-1]],
node_color='red', node_size=600, label='终点')
nx.draw_networkx_nodes(G, pos, nodelist=path[1:-1],
node_color='yellow', node_size=500, label='路径')
nx.draw_networkx_edges(G, pos, edgelist=path_edges,
edge_color='red', width=2)
plt.title(title)
plt.legend()
plt.show()
# 可视化结果
path, cost = networkx_astar(G, start, goal)
visualize_path(G, path, f"A* Path (Cost: {cost})")
完整的实用案例
class AStarPathFinder:
def __init__(self, grid_size=5, obstacles=None):
self.grid_size = grid_size
self.obstacles = obstacles or []
self.graph = self._create_graph()
def _create_graph(self):
G = nx.Graph()
rows, cols = self.grid_size, self.grid_size
# 添加节点
for i in range(rows):
for j in range(cols):
if (i, j) not in self.obstacles:
G.add_node((i, j), pos=(j, -i))
# 添加边（允许对角线移动）
directions = [(0, 1), (1, 0), (0, -1), (-1, 0), # 四方向
(1, 1), (1, -1), (-1, 1), (-1, -1)] # 对角线
for i in range(rows):
for j in range(cols):
if (i, j) in self.obstacles:
continue
for di, dj in directions:
ni, nj = i + di, j + dj
if (0 <= ni < rows and 0 <= nj < cols and
(ni, nj) not in self.obstacles):
# 对角线移动成本更高
weight = 1.414 if abs(di) + abs(dj) == 2 else 1
G.add_edge((i, j), (ni, nj), weight=weight)
return G
def find_path(self, start, goal, heuristic='manhattan'):
if heuristic == 'manhattan':
h = manhattan_distance
elif heuristic == 'euclidean':
h = euclidean_distance
else:
h = manhattan_distance
try:
path = nx.astar_path(self.graph, start, goal, heuristic=h)
cost = nx.astar_path_length(self.graph, start, goal, heuristic=h)
return path, cost
except nx.NetworkXNoPath:
return None, float('inf')
def visualize(self, start, goal, path=None):
pos = nx.get_node_attributes(self.graph, 'pos')
plt.figure(figsize=(12, 8))
# 绘制基础图
nx.draw(self.graph, pos, with_labels=True,
node_color='lightblue', node_size=400,
font_size=6, edge_color='gray', alpha=0.5)
# 标记障碍物
if self.obstacles:
nx.draw_networkx_nodes(self.graph, pos,
nodelist=self.obstacles,
node_color='black', node_size=400)
# 标记起点和终点
nx.draw_networkx_nodes(self.graph, pos, nodelist=[start],
node_color='green', node_size=600, label='起点')
nx.draw_networkx_nodes(self.graph, pos, nodelist=[goal],
node_color='red', node_size=600, label='终点')
# 绘制路径
if path:
path_edges = list(zip(path[:-1], path[1:]))
nx.draw_networkx_nodes(self.graph, pos, nodelist=path[1:-1],
node_color='yellow', node_size=400)
nx.draw_networkx_edges(self.graph, pos, edgelist=path_edges,
edge_color='red', width=3)
plt.title(f"A* Path Finding\nStart: {start}, Goal: {goal}")
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()
# 使用示例
finder = AStarPathFinder(grid_size=7, obstacles=[(2,2), (2,3), (2,4), (3,2), (4,2)])
start = (0, 0)
goal = (6, 6)
path, cost = finder.find_path(start, goal, heuristic='manhattan')
print(f"找到路径: {path}")
print(f"路径成本: {cost:.2f}")
finder.visualize(start, goal, path)
性能比较
import time
def compare_algorithms():
# 创建更大规模的图
G = create_grid_graph(20, 20)
start = (0, 0)
goal = (19, 19)
# 测试自定义实现
start_time = time.time()
path1, cost1 = astar_algorithm(G, start, goal, manhattan_distance)
custom_time = time.time() - start_time
# 测试NetworkX实现
start_time = time.time()
path2, cost2 = networkx_astar(G, start, goal, manhattan_distance)
networkx_time = time.time() - start_time
print(f"自定义A*: 路径长度={len(path1)}, 成本={cost1}, 时间={custom_time:.4f}s")
print(f"NetworkX A*: 路径长度={len(path2)}, 成本={cost2}, 时间={networkx_time:.4f}s")
print(f"路径是否一致: {path1 == path2}")
compare_algorithms()
关键知识点总结
-
核心概念：
g(n): 从起点到当前节点的实际成本h(n): 启发式估计成本f(n) = g(n) + h(n): 总成本函数
-
启发式函数：
- 曼哈顿距离（适用于四方向移动）
- 欧几里得距离（适用于对角线移动）
-
NetworkX优势：
- 内置
astar_path()和astar_path_length() - 自动处理图搜索
- 支持自定义启发式函数
- 内置
A*算法在处理路径规划、游戏AI、机器人导航等场景非常实用,NetworkX使其实现更加简洁高效。
