前言：

有一次学长在电话面试，碰巧我在周围的课桌上刷题，然后就（偷）听到了面试的内容。。。
  ~  
记忆比较深的就是面试官特意问了 A ∗ A^* A∗算法（那么多算法里偏偏挑了 A ∗ A^* A∗，一定是特别的缘分）；
  ~  
这位学长是ACM队里的大佬，现在已经保研，但是被问到 A ∗ A^* A∗算法的时候也楞了一下，
  ~  
毕竟这个算法接触的比较少，我之前也是大概懂个原理，没写过代码；
  ~  
碰巧这几天做了个 A ∗ A^* A∗的题，简单记录一下。

一、启发式搜索（可略过）

1.1 介绍

启发式搜索要用到问题自身的某些特性信息，以指导搜索朝着最有希望的方向前进。由于这种搜索针对性较强，因而原则上只需要搜索问题的部分状态空间，效率较高。

1.2 启发性信息

可用于指导搜索过程，且与具体问题求解有关的控制性信息。

1. 陈述性启发信息：一般被用于更准确、更精练地描述状态，使问题的状态空间缩小，如待求问题的特定状况等属于此类信息。
2. 过程性启发信息：一般被用于构造操作算子，使操作算子少而精，如一些规律性知识属于此类信息。
3. 控制性启发信息：它是关于表示控制策略方面的知识，包括协调整个问题求解过程中所使用的各种处理方法、搜索策略、控制结构等有关的知识。

1.3 估价函数

用于估价节点重要性或“有希望”程度的函数称为估价函数。

其一般形式为：
f ( x ) = g ( x ) + h ( x ) f(x)=g(x)+h(x) f(x)=g(x)+h(x)

• g ( x ) g(x) g(x)为从初始节点 S 0 S_0 S0​ 到节点 x x x 已经实际付出的代价；
• h ( x ) h(x) h(x) 是从节点 x x x 到目标节点 S S S 的最优路径的估计代价，它体现了问题的启发性信息，其形式要根据问题的特性确定。它可以是节点 x x x 到目标节点的距离或差异，可以是x格局的得分，也可以是节点 x x x 处于最优路径上的概率等等。 h ( x ) h (x) h(x) 称为启发函数。
• 估价函数 f(z) 表示从初始节点经过节点 Z 到达目标节点的最优路径的代价估计值，它的作用是估价 OPEN 表中各节点的重要程度，决定它们在 OPEN 表中的次序。

1.4 启发式算法

• 局部择优搜索
• 全局择优搜索
• A ∗ A^* A∗算法（最佳图搜索算法）

二、 A ∗ A^* A∗算法

A ∗ A^* A∗算法，又称最佳图搜索算法，是一种启发式搜索算法，是一种非常有效的搜索最短路径的算法。

左图是我们的 A ∗ A^* A∗算法，右图是 D i j k s t r a Dijkstra Dijkstra最短路算法，可以看到， A ∗ A^* A∗算法在搜索路径上是比 D i j k s t r a Dijkstra Dijkstra聪明得多的，那么这么高效的搜索算法是如何实现的呢？

2.1 算法分析

OPEN表：用于存放刚生成的节点，这些节点也是待扩展的，所以OPEN表也称为未扩展节点表；
CLOSED表：CLOSED表则是用来存放将要扩展或已经扩展的节点，所以它被称为已扩展节点表。

A ∗ A^* A∗算法也是从一般搜索过程中推广来的，对一般搜索过程作如下限制，即可成为 A ∗ A^* A∗算法：

1. 把OPEN表中的节点按估价函数 f ( x ) = g ( x ) + h ( x ) f(x)=g(x)+h(x) f(x)=g(x)+h(x)的从小到大进行排序；（代码中，我们使用了一个优先队列）
2. 代价函数 g ( x ) g(x) g(x)是对 g ∗ ( x ) g^*(x) g∗(x)的估计， g ∗ ( x ) > 0 g^*(x)>0 g∗(x)>0。 g ∗ ( x ) g^*(x) g∗(x)是从初始节点 S 0 S_0 S0​到节点 x x x的最小代价；
3. 启发函数 h ( x ) h(x) h(x)是 h ∗ ( x ) h^*(x) h∗(x)的下界，即对所有的 x x x均有： h ( x ) ≤ h ∗ ( x ) h(x) \leq h^*(x) h(x)≤h∗(x) 。 h ∗ ( x ) h^*(x) h∗(x)是从节点 x x x到目标节点的最小代价，若有多个目标节点，则为其中最小的一个。

        ~~~~~~~         通过 x x x的最佳路径： f ∗ ( x ) = g ∗ ( x ) + h ∗ ( x ) f^*(x)=g^*(x)+h^*(x) f∗(x)=g∗(x)+h∗(x)最小的路径，即从 S S S出发，通过节点 x x x的，到达目标节点的代价和最小的路径。

        ~~~~~~~         g ( x ) g(x) g(x)比较容易得到，它实际上就是从初始节点 S 0 S_0 S0​到节点x的路径代价，恒有 g ( x ) ≥ g ∗ ( x ) g(x)\geq g^*(x) g(x)≥g∗(x) ，而且在算法执行过程中随着更多搜索信息的获得， g ( x ) g(x) g(x)的值呈下降的趋势，也就是说 g ( x ) g(x) g(x)会慢慢接近最优距离 g ∗ ( x ) g^*(x) g∗(x)。

        ~~~~~~~         h ( x ) h(x) h(x)的确定依赖于具体问题领域的启发性信息，其中 h ( x ) ≤ h ∗ ( x ) h(x) \leq h^*(x) h(x)≤h∗(x)，这个限制条件是十分重要的，它可保证 A ∗ A^* A∗算法能找到最优解。
        ~~~~~~~        而且对于 h ( x ) h(x) h(x)的不同选择，可能会直接影响到算法的效率，举个例子，如果我们设 h ( x ) = 0 h(x)=0 h(x)=0，肯定满足了 h ( x ) ≤ h ∗ ( x ) h(x) \leq h^*(x) h(x)≤h∗(x)，那么 f ( x ) = g ( x ) f(x)=g(x) f(x)=g(x)，只根据源点到当前点 x x x的距离选择下一个点，这其实就是BFS（广度优先搜索）了，算法效率肯定很低。所以 h ( x ) h(x) h(x)的选择是非常重要的。

        ~~~~~~~         g ∗ ( x ) g^*(x) g∗(x)和 h ∗ ( x ) h^*(x) h∗(x)应该是最理想状态下的启发函数，但是我们人为设置的 g ( x ) g(x) g(x)和 h ( x ) h(x) h(x)可能并不是非常理想，但是只要满足 g ( x ) ≥ g ∗ ( x ) g(x)\geq g^*(x) g(x)≥g∗(x)和 h ( x ) ≤ h ∗ ( x ) h(x) \leq h^*(x) h(x)≤h∗(x)这两个条件就行。

2.2 算法流程

1. 把起点 S S S放入 O P E N OPEN OPEN表，记 f ( S ) = 0 + h ( S ) f(S)=0+h(S) f(S)=0+h(S)，令 C L O S E D CLOSED CLOSED为空表。
2. 重复下列过程，直至找到目标节点止。若 O P E N OPEN OPEN为空表还未找到目标节点，则宣告失败。
3. 选取 O P E N OPEN OPEN表中未扩展过的具有最小 f f f 值的节点为最佳节点 B E S T N O D E BESTNODE BESTNODE，并把它放入 C L O S E D CLOSED CLOSED表。
4. 若 B E S T N O D E BESTNODE BESTNODE为一目标节点，则成功求得一解。
5. 若 B E S T N O D E BESTNODE BESTNODE不是目标节点，则扩展之，产生后继节点集 S U C C S S O R i ∣ i = 1 , 2 , … , n {SUCCSSOR_i|i=1,2,…,n} SUCCSSORi​∣i=1,2,…,n。
6. 对每个 S U C C S S O R i SUCCSSORi SUCCSSORi进行下列过程：
   1. 建立从 S U C C S S O R i SUCCSSOR_i SUCCSSORi​返回 B E S T N O D E BESTNODE BESTNODE的指针。
   2. 计算 g ( S U C S S O R i ) = g ( B E S T N O D E ) + c ( B E S N O D E , S U C S S O R i ) g(SUCSSOR_i)=g(BESTNODE)+ c(BESNODE,SUCSSOR_i) g(SUCSSORi​)=g(BESTNODE)+c(BESNODE,SUCSSORi​)。
   3. 如果 S U C C S S O R i SUCCSSOR_i SUCCSSORi​在 O P E N OPEN OPEN表中,则比较新旧路径代价。如果从 S 0 S0 S0到 S U C S S O R i SUCSSOR_i SUCSSORi​的新路径 ＜ ＜ ＜旧路径,则重新确定 S U C S S O R i SUCSSOR_i SUCSSORi​的父辈节点为 B E S T N O D E BESTNODE BESTNODE，记下较小代价 g ( S U C S S O R i ) g(SUCSSOR_i) g(SUCSSORi​),并修正 f ( S U C S S O R i ) f(SUCSSOR_i) f(SUCSSORi​)值。
   4. 如果 S U C C S S O R i SUCCSSOR_i SUCCSSORi​不在 O P E N OPEN OPEN表中，在 C L O S E CLOSE CLOSE表中,且从S0到SUCSSOR_i的新路径 < C L O S E D <CLOSED <CLOSED表中的 g ( S U C C S S O R i ) g(SUCCSSOR_i) g(SUCCSSORi​),则更新 C L O S E D CLOSED CLOSED表中的估价函数值并从 C L O S E D CLOSED CLOSED表中移出节点 S U C C S S O R i SUCCSSOR_i SUCCSSORi​放入 O P E N OPEN OPEN表中。
   5. 如果 S U C C S S O R i SUCCSSOR_i SUCCSSORi​不在 O P E N OPEN OPEN表中也不在 C L O S E D CLOSED CLOSED表中时，求出 S U C C S S O R i SUCCSSOR_i SUCCSSORi​的估价函数值并将其插入 O P E N OPEN OPEN表中。
7. 按照估价函数值将 O P E N OPEN OPEN表中的节点排序；
8. 循环

2.3 A*算法的特性

• A*算法可纳性
          ~~~~~~~        对于可解状态空间图（即从初始节点到目标节点有路径存在）来说，如果一个搜索算法能在有限步内终止，并且能找到最优解，则称该搜索算法是可纳的。 A ∗ A^* A∗算法是可纳的，即它能在有限步内终止并找到最优解。
• A*算法的最优性
          A ∗ ~~~~~~~A^*        A∗算法的效率在很大程度上取决于 h ( x ) h(x) h(x)，在满足 h ( x ) ≤ h ∗ ( x ) h(x)≤h^*(x) h(x)≤h∗(x)的前提下， h ( x ) h(x) h(x)的值越大越好。 h ( x ) h(x) h(x)的值越大，表明它携带的启发性信息越多，搜索时扩展的节点数越少，搜索的效率越高。
• h(x)的单调性限制
          ~~~~~~~        在 A ∗ A^* A∗算法中，每当要扩展一个节点时都要先检查其子节点是否已在OPEN 表或CLOSED表中，有时还需要调整指向父节点的指针，这就增加了搜索的代价。如果对启发函数 h ( x ) h(x) h(x)加上单调性限制，就可减少检查及调整的工作量，从而减少搜索代价。

2.4 核心代码

当时是做的一个国外学校的课程作业，整个项目涉及了DFS、BFS、A star、图等价等知识，这里把关于A star算法的相关内容摘录出来了。核心代码是照着上面的思路写的，理解思路后，代码是很容易懂的。

    def a_star(self, start_id, target_id):
        dict_g = {start_id:0}
        def calc_hx(current_id, target_id): # 计算h(x)，这里的h(x)是用了最简单的欧几里得距离
            return self.get_vertex(current_id).euclidean_distance(self.get_vertex(target_id))
        def calc_gx(parent_id, current_id, current_weight):
            return dict_g[parent_id] + current_weight
        if (start_id not in self.vertices) or (target_id not in self.vertices):
            return ([], 0)
        open_list = AStarPriorityQueue() # 类似于优先队列，具体结构在下面的完整代码中可以找到
        close_list = []
        parent = {}
        open_list.push(0, self.get_vertex(start_id))
        while not open_list.empty():
            _, current_v = open_list.pop()
             close_list.append(current_v)
            if current_v.id == target_id:
                break
            for edge in current_v.adj.items():
                if edge[0] not in self.vertices:
                    continue
                vertex_hx = calc_hx(edge[0], target_id)
                vertex_gx = calc_gx(current_v.id, edge[0], edge[1])
                vertex_fx = vertex_hx + vertex_gx
                edge_v = self.get_vertex(edge[0])
                # 下面几个if else 是算法的核心部分
                if edge[0] in open_list.locator():
                    if vertex_gx < dict_g[edge[0]]:
                        open_list.update(vertex_fx, edge_v)
                        dict_g[edge[0]] = vertex_gx
                        parent[edge[0]] = current_v.id
                elif edge_v in close_list:
                    if vertex_gx < dict_g[edge[0]]:
                        open_list.update(vertex_fx, edge_v)
                        dict_g[edge[0]] = vertex_gx
                        parent[edge[0]] = current_v.id
                        open_list.push(vertex_fx, edge[0])
                else:
                    open_list.push(vertex_fx, edge_v)
                    dict_g[edge[0]] = vertex_gx
                    parent[edge[0]] = current_v.id
        if target_id in dict_g:
            res = [target_id]
            now_id = target_id
            while now_id != start_id:
                res.append(parent[now_id])
                now_id = parent[now_id]
            res.reverse()
            return (res, dict_g[target_id])
        return ([], 0)

2.5 完整代码

2.5 完整代码 + 2.6 测试 两大部分可直接运行

import random
import queue, heapq, math, itertools
from collections import OrderedDict
import numpy as np

class Vertex:
    """ Class representing a Vertex object within a Graph """
    __slots__ = ['id', 'adj', 'visited', 'x', 'y']

    def __init__(self, idx, x=0, y=0):
        """ Initializes a Vertex :param idx: A unique string identifier used for hashing the vertex :param x: The x coordinate of this vertex (used in a_star) :param y: The y coordinate of this vertex (used in a_star) """
        self.id = idx
        self.adj = OrderedDict()  # dictionary {id : weight} of outgoing edges
        self.visited = False      # Boolean flag used in search algorithms
        self.x = x
        self.y = y

    def visit(self):
        self.visited = True

    def reset(self):
        self.visited = False

    def euclidean_distance(self, other):
        return math.sqrt((self.x-other.x)*(self.x-other.x)+(self.y-other.y)*(self.y-other.y))

class Graph:
    """ Class implementing the Graph ADT using an Adjacency Map structure """
    __slots__ = ['size', 'vertices', 'plot_show', 'plot_delay']

    def __init__(self, plt_show=False):
        """ Instantiates a Graph class instance :param: plt_show : if true, render plot when plot() is called; else, ignore calls to plot() """
        self.size = 0
        self.vertices = OrderedDict()
        self.plot_show = plt_show
        self.plot_delay = 0.2

    def reset_vertices(self):
        for k, v in self.vertices.items():
            v.reset()

    def get_vertex(self, vertex_id):
        if vertex_id in self.vertices:
            return self.vertices[vertex_id]
        else:
            return None

    def add_to_graph(self, start_id, dest_id=None, weight=0):
        if start_id == None:
            return
        if start_id not in self.vertices:
            self.vertices[start_id] = Vertex(start_id)
            self.size += 1
        if dest_id != None:
            if dest_id not in self.vertices:
                self.vertices[dest_id] = Vertex(dest_id)
                self.size += 1
            self.vertices[start_id].adj[dest_id] = weight

    def a_star(self, start_id, target_id):
        dict_g = {start_id:0}
        def calc_hx(current_id, target_id):
            return self.get_vertex(current_id).euclidean_distance(self.get_vertex(target_id))
        def calc_gx(parent_id, current_id, current_weight):
            return dict_g[parent_id] + current_weight
        if (start_id not in self.vertices) or (target_id not in self.vertices):
            return ([], 0)
        open_list = AStarPriorityQueue()
        close_list = []
        parent = {}
        open_list.push(0, self.get_vertex(start_id))
        while not open_list.empty():
            _, current_v = open_list.pop()
            close_list.append(current_v)
            if current_v.id == target_id:
                break
            for edge in current_v.adj.items():
                if edge[0] not in self.vertices:
                    continue
                vertex_hx = calc_hx(edge[0], target_id)
                vertex_gx = calc_gx(current_v.id, edge[0], edge[1])
                vertex_fx = vertex_hx + vertex_gx
                edge_v = self.get_vertex(edge[0])
                if edge[0] in open_list.locator():
                    if vertex_gx < dict_g[edge[0]]:
                        open_list.update(vertex_fx, edge_v)
                        dict_g[edge[0]] = vertex_gx
                        parent[edge[0]] = current_v.id
                elif edge_v in close_list:
                    if vertex_gx < dict_g[edge[0]]:
                        open_list.update(vertex_fx, edge_v)
                        dict_g[edge[0]] = vertex_gx
                        parent[edge[0]] = current_v.id
                        open_list.push(vertex_fx, edge[0])
                else:
                    open_list.push(vertex_fx, edge_v)
                    dict_g[edge[0]] = vertex_gx
                    parent[edge[0]] = current_v.id
        if target_id in dict_g:
            res = [target_id]
            now_id = target_id
            while now_id != start_id:
                res.append(parent[now_id])
                now_id = parent[now_id]
            res.reverse()
            return (res, dict_g[target_id])
        return ([], 0)

class AStarPriorityQueue:
    __slots__ = ['__data', '__locator', '__counter']

    def __init__(self):
        """ Construct an AStarPriorityQueue object """
        self.__data = []                        # underlying data list of priority queue
        self.__locator = {}                     # dictionary to locate vertices within priority queue
        self.__counter = itertools.count()      # used to break ties in prioritization

    def locator(self):
        return self.__locator
    def empty(self):
        """ Determine whether priority queue is empty :return: True if queue is empty, else false """
        return len(self.__data) == 0

    def push(self, priority, vertex):
        """ Push a vertex onto the priority queue with a given priority :param priority: priority key upon which to order vertex :param vertex: Vertex object to be stored in the priority queue :return: None """
        count = next(self.__counter)
        # list is stored by reference, so updating will update all refs
        node = [priority, count, vertex]
        self.__locator[vertex.id] = node
        heapq.heappush(self.__data, node)

    def pop(self):
        """ Remove and return the (priority, vertex) tuple with lowest priority key :return: (priority, vertex) tuple where priority is key, and vertex is Vertex object stored in priority queue """
        vertex = None
        while vertex is None:
            # keep popping until we have valid entry
            priority, count, vertex = heapq.heappop(self.__data)
        del self.__locator[vertex.id]                   # remove from locator dict
        vertex.visit()                  # indicate that this vertex was visited
        return priority, vertex

    def update(self, new_priority, vertex):
        """ Update given Vertex object in the priority queue to have new priority :param new_priority: new priority on which to order vertex :param vertex: Vertex object for which priority is to be updated :return: None """
        node = self.__locator.pop(vertex.id)    # delete from dictionary
        node[-1] = None                         # invalidate old node
        self.push(new_priority, vertex)         # push new node

2.6 测试

def test_a_star():
    graph = Graph()
    vertices = [Vertex('A', 0, 0), Vertex('B', 2, 0), Vertex('C', 4, 0), Vertex('D', 6, 0), Vertex('E', 9, 0),
                Vertex('F', 12, 0), Vertex('G', 2, 5), Vertex('H', 6, 4), Vertex('I', 12, 4), Vertex('J', 5, 9),
                Vertex('K', 8, 8), Vertex('L', 12, 8), Vertex('M', 8, 10), Vertex('Breslin Center', 0, 2),
                Vertex('Spartan Stadium', 4, 2), Vertex('Wells Hall', 9, 2),
                Vertex('Engineering Building', 9, -2),
                Vertex('Library', 7, 6), Vertex('Union', 8, 11), Vertex('The Rock', 14, 8)]
    for vertex in vertices:
        graph.vertices[vertex.id] = vertex
    edges = [('A', 'B', 8), ('B', 'C', 8), ('C', 'D', 8), ('D', 'E', 12), ('E', 'F', 12), ('B', 'G', 5),
                 ('D', 'H', 4), ('F', 'I', 16),
                 ('G', 'H', 5), ('H', 'I', 6), ('G', 'J', 5),
                 ('I', 'L', 16), ('J', 'K', 4), ('K', 'L', 4),
                 ('J', 'M', 4), ('M', 'L', 4),
                 ('Breslin Center', 'A', 0), ('Spartan Stadium', 'C', 0), ('Wells Hall', 'E', 0),
                 ('Engineering Building', 'E', 0),
                 ('Library', 'K', 0), ('Union', 'M', 0), ('The Rock', 'L', 0)]
    for edge in edges:
        # add edge in both directions
        graph.add_to_graph(edge[0], edge[1], edge[2])
        graph.add_to_graph(edge[1], edge[0], edge[2])
        # test Breslin to Union shortest path
    subject = graph.a_star('Breslin Center', 'Union')
    path = ['Breslin Center', 'A', 'B', 'G', 'J', 'M', 'Union']
    assert subject == (path, 22)
    graph.reset_vertices()
    path.reverse()
    subject = graph.a_star('Union', 'Breslin Center')
    assert subject == (path, 22)
    graph.reset_vertices()

    # test Breslin to EB shortest path - bypass slow Shaw Ln
    subject = graph.a_star('Breslin Center', 'Engineering Building')
    path = ['Breslin Center', 'A', 'B', 'G', 'H', 'D', 'E', 'Engineering Building']
    assert subject == (path, 34)
    graph.reset_vertices()
    path.reverse()
    subject = graph.a_star('Engineering Building', 'Breslin Center')
    assert subject == (path, 34)
    graph.reset_vertices()
    # test EB to The Rock shortest path - bypass slow Farm Ln
    subject = graph.a_star('Engineering Building', 'The Rock')
    path = ['Engineering Building', 'E', 'D', 'H', 'G', 'J', 'K', 'L', 'The Rock']
    assert subject == (path, 34)
    graph.reset_vertices()
    path.reverse()
    subject = graph.a_star('The Rock', 'Engineering Building')
    assert subject == (path, 34)
    graph.reset_vertices()

    # test Union to Library - despite equal path lengths, A* heuristic will direct search to left
    subject = graph.a_star('Union', 'Library')
    path = ['Union', 'M', 'J', 'K', 'Library']
    assert subject == (path, 8)
    graph.reset_vertices()
    path.reverse()
    subject = graph.a_star('Library', 'Union')
    assert subject == (path, 8)
    graph.reset_vertices()
    print("ok")
test_a_star()