前言：

有一次学长在电话面试，碰巧我在周围的课桌上刷题，然后就（偷）听到了面试的内容。。。

$~$

 
记忆比较深的就是面试官特意问了
$A^*$

A∗算法（那么多算法里偏偏挑了
$A^*$

A∗，一定是特别的缘分）；

$~$

 
这位学长是ACM队里的大佬，现在已经保研，但是被问到
$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)

• 
  $g(x)$

  g(x)为从初始节点
  $S_0$

  S0​ 到节点
  $x$

  x 已经实际付出的代价；

• 
  $h(x)$

  h(x) 是从节点
  $x$

  x 到目标节点
  $S$

  S 的最优路径的估计代价，它体现了问题的启发性信息，其形式要根据问题的特性确定。它可以是节点
  $x$

  x 到目标节点的距离或差异，可以是x格局的得分，也可以是节点
  $x$

  x 处于最优路径上的概率等等。
  $h (x)$

  h(x) 称为启发函数。

• 估价函数 f(z) 表示从初始节点经过节点 Z 到达目标节点的最优路径的代价估计值，它的作用是估价 OPEN 表中各节点的重要程度，决定它们在 OPEN 表中的次序。

1.4 启发式算法

• 局部择优搜索
• 全局择优搜索
• 
  $A^*$

  A∗算法（最佳图搜索算法）

二、
$A^*$

A∗算法

$A^*$

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

左图是我们的
$A^*$

A∗算法，右图是
$Dijkstra$

Dijkstra最短路算法，可以看到，
$A^*$

A∗算法在搜索路径上是比
$Dijkstra$

Dijkstra聪明得多的，那么这么高效的搜索算法是如何实现的呢？

2.1 算法分析

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

$A^*$

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

A∗算法：

1. 把OPEN表中的节点按估价函数
   $f(x)=g(x)+h(x)$

   f(x)=g(x)+h(x)的从小到大进行排序；（代码中，我们使用了一个优先队列）

2. 代价函数
   $g(x)$

   g(x)是对
   $g^*(x)$

   g∗(x)的估计，
   $g^*(x)>0$

   g∗(x)>0。
   $g^*(x)$

   g∗(x)是从初始节点
   $S_0$

   S0​到节点
   $x$

   x的最小代价；

3. 启发函数
   $h(x)$

   h(x)是
   $h^*(x)$

   h∗(x)的下界，即对所有的
   $x$

   x均有：
   $h(x) leq h^*(x)$

   h(x)≤h∗(x) 。
   $h^*(x)$

   h∗(x)是从节点
   $x$

   x到目标节点的最小代价，若有多个目标节点，则为其中最小的一个。

$~~~~~~~$

        通过
$x$

x的最佳路径：
$f^*(x)=g^*(x)+h^*(x)$

f∗(x)=g∗(x)+h∗(x)最小的路径，即从
$S$

S出发，通过节点
$x$

x的，到达目标节点的代价和最小的路径。

$~~~~~~~$

       
$g(x)$

g(x)比较容易得到，它实际上就是从初始节点
$S_0$

S0​到节点x的路径代价，恒有
$g(x)geq g^*(x)$

g(x)≥g∗(x) ，而且在算法执行过程中随着更多搜索信息的获得，
$g(x)$

g(x)的值呈下降的趋势，也就是说
$g(x)$

g(x)会慢慢接近最优距离
$g^*(x)$

g∗(x)。

$~~~~~~~$

       
$h(x)$

h(x)的确定依赖于具体问题领域的启发性信息，其中
$h(x) leq h^*(x)$

h(x)≤h∗(x)，这个限制条件是十分重要的，它可保证
$A^*$

A∗算法能找到最优解。

$~~~~~~~$

       而且对于
$h(x)$

h(x)的不同选择，可能会直接影响到算法的效率，举个例子，如果我们设
$h(x)=0$

h(x)=0，肯定满足了
$h(x) leq h^*(x)$

h(x)≤h∗(x)，那么
$f(x)=g(x)$

f(x)=g(x)，只根据源点到当前点
$x$

x的距离选择下一个点，这其实就是BFS（广度优先搜索）了，算法效率肯定很低。所以
$h(x)$

h(x)的选择是非常重要的。

$~~~~~~~$

       
$g^*(x)$

g∗(x)和
$h^*(x)$

h∗(x)应该是最理想状态下的启发函数，但是我们人为设置的
$g(x)$

g(x)和
$h(x)$

h(x)可能并不是非常理想，但是只要满足
$g(x)geq g^*(x)$

g(x)≥g∗(x)和
$h(x) leq h^*(x)$

h(x)≤h∗(x)这两个条件就行。

2.2 算法流程

1. 把起点
   $S$

   S放入
   $OPEN$

   OPEN表，记
   $f(S)=0+h(S)$

   f(S)=0+h(S)，令
   $CLOSED$

   CLOSED为空表。

2. 重复下列过程，直至找到目标节点止。若
   $OPEN$

   OPEN为空表还未找到目标节点，则宣告失败。

3. 选取
   $OPEN$

   OPEN表中未扩展过的具有最小
   $f$

   f 值的节点为最佳节点
   $BESTNODE$

   BESTNODE，并把它放入
   $CLOSED$

   CLOSED表。

4. 若
   $BESTNODE$

   BESTNODE为一目标节点，则成功求得一解。

5. 若
   $BESTNODE$

   BESTNODE不是目标节点，则扩展之，产生后继节点集
   ${SUCCSSOR_i|i=1,2,…,n}$

   SUCCSSORi​∣i=1,2,…,n。

6. 对每个
   $SUCCSSORi$

   SUCCSSORi进行下列过程：

   1. 建立从
      $SUCCSSOR_i$

      SUCCSSORi​返回
      $BESTNODE$

      BESTNODE的指针。

   2. 计算
      $g(SUCSSOR_i)=g(BESTNODE)+ c(BESNODE,SUCSSOR_i)$

      g(SUCSSORi​)=g(BESTNODE)+c(BESNODE,SUCSSORi​)。

   3. 如果
      $SUCCSSOR_i$

      SUCCSSORi​在
      $OPEN$

      OPEN表中,则比较新旧路径代价。如果从
      $S0$

      S0到
      $SUCSSOR_i$

      SUCSSORi​的新路径
      $＜$

      ＜旧路径,则重新确定
      $SUCSSOR_i$

      SUCSSORi​的父辈节点为
      $BESTNODE$

      BESTNODE，记下较小代价
      $g(SUCSSOR_i)$

      g(SUCSSORi​),并修正
      $f(SUCSSOR_i)$

      f(SUCSSORi​)值。

   4. 如果
      $SUCCSSOR_i$

      SUCCSSORi​不在
      $OPEN$

      OPEN表中，在
      $CLOSE$

      CLOSE表中,且从S0到SUCSSOR_i的新路径
      $<CLOSED$

      <CLOSED表中的
      $g(SUCCSSOR_i)$

      g(SUCCSSORi​),则更新
      $CLOSED$

      CLOSED表中的估价函数值并从
      $CLOSED$

      CLOSED表中移出节点
      $SUCCSSOR_i$

      SUCCSSORi​放入
      $OPEN$

      OPEN表中。

   5. 如果
      $SUCCSSOR_i$

      SUCCSSORi​不在
      $OPEN$

      OPEN表中也不在
      $CLOSED$

      CLOSED表中时，求出
      $SUCCSSOR_i$

      SUCCSSORi​的估价函数值并将其插入
      $OPEN$

      OPEN表中。

7. 按照估价函数值将
   $OPEN$

   OPEN表中的节点排序；

8. 循环

2.3 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)的单调性限制
  
  $~~~~~~~$

         在
  $A^*$

  A∗算法中，每当要扩展一个节点时都要先检查其子节点是否已在OPEN 表或CLOSED表中，有时还需要调整指向父节点的指针，这就增加了搜索的代价。如果对启发函数
  $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()