加速 Python 中的 A* 实现

Question

A* 算法是一种类似于 Dijkstra 算法的寻路算法，其工作原理是访问节点（使用启发式方法来决定接下来访问哪个节点），并将该节点与已访问且处于封闭状态的节点进行比较列表。

在我的实现中，随着封闭列表大小的增加，每秒访问的节点数急剧减少。虽然最初，该算法访问大约 3,000 个节点/秒，但随着封闭列表增长 > 10,000 个节点，这将减少到小于 50 个节点/秒。唯一在计算上变得更昂贵的是新节点与开放列表和封闭列表的比较，并将新节点存储在封闭列表中。

因此，我认为通过以更有效的方式存储封闭列表可以显着提高性能！

以下是我实施的一些摘录。一是Node类，用于定义所有Node：

class Node:
    """
    A node class for A* Pathfinding
    """

    def __init__(self, parent=None, position=None):
        self.parent = parent
        self.position = position

        self.g = 0 # g = actual cost of reaching this node
        self.h = 0 # h = heuristic, used for determining which node to visit next
        self.f = 0 # f = g + h

    def __eq__(self, other):
        return self.position == other.position

    # defining less than for purposes of heap queue
    def __lt__(self, other):
        return self.f < other.f

    # defining greater than for purposes of heap queue
    def __gt__(self, other):
        return self.f > other.f

我使用堆队列来存储打开列表，因为我认为这可以提高速度。然而，它只是略微（±5%）。

下面是我的 A* 实现，精简后只包含相关操作：

def find_a_star_path(self, current_pos, target_pos):
    # Initialize start- and end-nodes with zero cost
    start_node = self.Node(None, current_pos)
    start_node.g = start_node.h = start_node.f = 0.0

    end_node = self.Node(None, target_pos)
    end_node.g = end_node.h = end_node.f = 0.0

    # Initialize open- and closed list
    open_list = []
    closed_list = []

    # Heapify the open_list and Add the start node
    heapq.heapify(open_list)
    heapq.heappush(open_list, start_node)

    # As long as there are "open" nodes, we continue A*.
    while len(open_list) > 0:
        # Find node with the lowest cost F, this is visited next
        current_node = heapq.heappop(open_list)
        closed_list.append(current_node)

        if current_node == end_node:
            # if current_node = end_node, the process is finished.

        # Some code that finds all possible next nodes from the next node
        # This next node is called child
        # child.g, child.h and child.f are calculated

        # Now check if the new node is better than another node with the same position but a different parent.

        filtered_open_nodes = (open_node for open_node in open_list if child == open_node)
        open_node = next(filtered_open_nodes, None)

        while open_node:
            if child.f > open_node.f:
                add_to_open = False
                break
            else:
                # The new node is better than the other path to this node, so remove it.
                open_list.remove(open_node)
                open_node = next(filtered_open_nodes, None)

        if add_to_open == True:
            heapq.heappush(open_list, child)

Answer 1

您的代码存在一些问题：

• 你从来没有真正使用过closed_list；您向其添加节点，但您从不检查 current_node 是否已关闭
• 您的closed_list 应该是set 用于O(1) 查找；但是，这意味着您要么只添加position，要么也实现Node.__hash__
• 通过执行open_list.remove(open_node)，您可能使由heapq 中的算法精心维护的堆不变量无效，这也可能导致运行时间更长或更糟，您的 A* 没有找到正确的结果
• 此处不相关，但Node.__eq__的实现应与__lt__和__gt__（通过不同属性比较）和__hash__（未实现）一致

有了这个，还有一些外观上的改变，例如使用any 检查open_list，您的代码可能如下所示：

def find_a_star_path(self, current_pos, target_pos):
    start_node = self.Node(None, current_pos)
    start_node.g = start_node.h = start_node.f = 0.0
    # no need for end node, just compare position

    open_list = [start_node] # no need to heapify list with just one element
    closed_set = set() # should be a set for O(1) "in" check

    while len(open_list) > 0:
        current_node = heapq.heappop(open_list)
        
        # checking the position here, alternatively implement Node.__hash__
        if current_node.position in closed_set:
            continue
        closed_set.add(current_node.position)

        if current_node.position == target_pos:
            # if current_node = end_node, the process is finished.

        for child in [code that finds all possible next nodes]:

            add_to_open = child.position not in closed_set and \
                not any(open_node.f <= child.f for open_node in open_list if open_node == child_node)

            if add_to_open:
                heapq.heappush(open_list, child)

但这可能仍然很慢，因为在每个步骤中对整个open_list 的线性扫描大大超过了 O(logn) 堆操作。通常，您可以不检查候选节点是否已经在open_list中，即删除and not any(...)检查。这可能会导致堆上的节点比必要的多，但这可能根本不是问题。一旦它们被弹出，它们就会被丢弃，因为它们已经在closed_set 中。 （实际上你可能会删除整个add_to_open 检查，但检查它们是否已经在closed_set 这里很便宜，所以为什么不呢。）

如果堆上的其他（可能是重复的）元素导致问题，您可以用 dict 映射位置（或节点，如果它们实现 __hash__）替换 open_list 的 O(n) 扫描) 到最小 f 值，提供 O(1) 查找，就像 closed_set：

    closed_set = set()
    open_dict = {start_node.position: start_node.f}

然后：

    for child in [code that finds all possible next nodes]:
        pos = child.position
        add_to_open = pos not in closed_set and \
                (pos not in open_dict or open_dict[pos] > child.f)

        if add_to_open:
            heapq.heappush(open_list, child)
            open_dict[pos] = child.f