python - Python 中的统一成本搜索

Question

我在 Python 中实现了一个简单的图形数据结构，结构如下。此处的代码只是为了阐明函数/变量的含义，但它们是不言自明的，因此您可以跳过阅读。

# Node data structure
class Node: 

    def __init__(self, label):        
        self.out_edges = []
        self.label = label
        self.is_goal = False


    def add_edge(self, node, weight = 0):          
        self.out_edges.append(Edge(node, weight))


# Edge data structure
class Edge:

    def __init__(self, node, weight = 0):          
        self.node = node
        self.weight = weight

    def to(self):                                  
        return self.node


# Graph data structure, utilises classes Node and Edge
class Graph:    

    def __init__(self):                             
        self.nodes = []

    # some other functions here populate the graph, and randomly select three goal nodes.

现在我正在尝试实现 uniform-cost search (即具有优先级队列的 BFS，保证最短路径)从给定节点 v 开始，并返回最短路径(以列表形式)到三个目标节点之一。 目标节点 是指属性 is_goal 设置为 true 的节点。

这是我的实现:

def ucs(G, v):
    visited = set()                  # set of visited nodes
    visited.add(v)                   # mark the starting vertex as visited
    q = queue.PriorityQueue()        # we store vertices in the (priority) queue as tuples with cumulative cost
    q.put((0, v))                    # add the starting node, this has zero *cumulative* cost   
    goal_node = None                 # this will be set as the goal node if one is found
    parents = {v:None}               # this dictionary contains the parent of each node, necessary for path construction

    while not q.empty():             # while the queue is nonempty
        dequeued_item = q.get()        
        current_node = dequeued_item[1]             # get node at top of queue
        current_node_priority = dequeued_item[0]    # get the cumulative priority for later

        if current_node.is_goal:                    # if the current node is the goal
            path_to_goal = [current_node]           # the path to the goal ends with the current node (obviously)
            prev_node = current_node                # set the previous node to be the current node (this will changed with each iteration)

            while prev_node != v:                   # go back up the path using parents, and add to path
                parent = parents[prev_node]
                path_to_goal.append(parent)   
                prev_node = parent

            path_to_goal.reverse()                  # reverse the path
            return path_to_goal                     # return it

        else:
            for edge in current_node.out_edges:     # otherwise, for each adjacent node
                child = edge.to()                   # (avoid calling .to() in future)

                if child not in visited:            # if it is not visited
                    visited.add(child)              # mark it as visited
                    parents[child] = current_node   # set the current node as the parent of child
                    q.put((current_node_priority + edge.weight, child)) # and enqueue it with *cumulative* priority

现在，经过大量测试并与其他算法进行比较后，这个实现似乎运行良好 - 直到我用这张图尝试它:

无论出于何种原因，ucs(G,v) 返回的路径 H -> I 成本为 0.87，而不是路径 H -> F -> I，成本为0.71(此路径是通过运行DFS获得的)。下图也给出了错误的路径:

算法给出了 G -> F 而不是 G -> E -> F，由 DFS 再次获得。在这些罕见的情况下，我能观察到的唯一模式是所选目标节点始终有一个循环。我不知道出了什么问题。任何提示将不胜感激。

Answer 1

通常对于搜索，我倾向于保留队列中节点部分的路径。这并不是真正的内存高效，但实现起来更便宜。

如果你想要父 map ，请记住只有当 child 在队列顶部时更新父 map 才是安全的。只有这样算法才能确定到当前节点的最短路径。

def ucs(G, v):
    visited = set()                  # set of visited nodes
    q = queue.PriorityQueue()        # we store vertices in the (priority) queue as tuples 
                                     # (f, n, path), with
                                     # f: the cumulative cost,
                                     # n: the current node,
                                     # path: the path that led to the expansion of the current node
    q.put((0, v, [v]))               # add the starting node, this has zero *cumulative* cost 
                                     # and it's path contains only itself.

    while not q.empty():             # while the queue is nonempty
        f, current_node, path = q.get()
        visited.add(current_node)    # mark node visited on expansion,
                                     # only now we know we are on the cheapest path to
                                     # the current node.

        if current_node.is_goal:     # if the current node is a goal
            return path              # return its path
        else:
            for edge in in current_node.out_edges:
                child = edge.to()
                if child not in visited:
                    q.put((current_node_priority + edge.weight, child, path + [child]))

注意:我还没有真正测试过这个，所以如果它不能立即起作用，请随时发表评论。