A星算法Python实现：移除Fringe中与Expanded重复的节点

我用Python实现A星（A-Star）算法解决八数码问题时，简单测试用例运行正常，但复杂用例如[6, 4, 0, 1, 5, 3, 7, 8, 2]会出现重复节点被扩展的问题。核心需求是：移除Fringe（待扩展节点列表）中与Expanded（已扩展节点列表）重复的元素，因为A星要求已扩展节点不能被Fringe节点再次访问。

原实现代码 #

from math import sqrt
import time
#fring- to keep the nodes 
#expand: the function to expand one node at a time 
    #heuristic: calculate the cheapest cost
    #f()- total cost 
    #h()- the number index to the goal 
#expanded_nodes- have all the visited 
startime = time.time_ns()
def astar(puzzle):
    #intializing the variables 
    cost = 0
    node = Node(puzzle,cost)
    loop = True
 
    #the possible nodes  
    fringe = []  
    #After expanding the list 
    expanded = []

    # maybe need it keep it for now 
    visit = set() #keep track of all the visit 
    
    #the goal state
    goal = [0,1,2,3,4,5,6,7,8] 
    
    #possible combination move
    possible_move = [[1,3],[0,2,4],[1,5],[0,4,6],[1,3,5,7],[2,4,8],[3,7],[4,6,8],[5,7]]
   
    #intialization of the first node 
    fringe = [node]
    #print('state:'+str(node.state))
    
    #start the loop 
    while loop:
        print('\nthe state of this game position is:\n ' + str(node.state) +"\n\n")
        #find the lowest cost in the fringe then expand the lowest cost node
        min_cost = fringe[0].cost 
        min_node = fringe[0]
        for node in fringe:
            if node.cost < min_cost:
                min_cost = node.cost
                min_node = node
        #append the node that was just expaneded into the expanded list but keep the list not expaneded in fringe 
        expanded.append(min_node)
        print('the min node '+str(min_node.state)+'\nthe cheapest cost: '+ str(min_cost) + '\n')
       
        #removes the min cost node from fringe  
        for node in fringe[:]:
            if node.state == min_node.state:
                fringe.remove(node)
               
        #when there is a solution in the expanded 
        for node in expanded:
            if node.state == goal:
                loop = False
                
        #checking the node in fringe 
        for node in fringe:
            print(node.state)
            
            
        #     key = tuple(node.state)
        # if key in visit:
        #     continue
        # visit.add(key)
        #traverse the nodes that was expanded and add the children to the fringe 
       # for node in expanded[:]:
        for node in expanded[:]:
            #append all the successor/children and set the children's parent to fringe 
            blank = node.state.index(8)
            print('the index of the blank is '+ str(blank))
            print('\n')
            possible_pos = possible_move[blank]
            print('possible pos '+ str(possible_pos))
                
            for i in possible_pos:
                #if node not in visit:
                    print('\n')
                    possible_sw = node.state[:]
                    print('index swap = '+ str(i))
                    possible_sw[blank] = possible_sw[i]
                    possible_sw[i] = 8
                    print('the child node is ' + str(possible_sw))
                    node.cost = manhattan(possible_sw, goal)
                    fringe.append(Node(possible_sw,node.cost,node))
                    print('the cost this node state: '+ str(node.cost)) 
   
        for node in expanded[:]:
            if node.cost > min_cost:
                expanded.pop(0)

    #finding the solution 
    solution = expanded
    move = 0
    while node.parent:
        solution.append(node.state.index(8))
        node = node.parent
        move += 1
    print('moves made '+ str(move))

    solution.reverse()
    print('moves list '+ str(solution))
    endtime = time.time_ns()
    executionTime = ( endtime - startime)
    print('Execution time in ns: ' + str(executionTime))
    
    
    return solution     

#Try the Manhattan Distance for moving only four direction up,down,left,right 
def manhattan(a, b):
        return sum(abs(val1-val2) for val1, val2 in zip(a,b))

class Node:
    def __init__(self,state,cost,parent = None):
        self.parent = parent
        self.state = state
        self.cost = cost
        self.children = []


#test case 
p = [0, 1, 2, 3, 4, 5, 6, 8, 7]
p = [0, 1, 2, 3, 8, 4, 6, 7, 5]
#p= [6, 4, 0, 1, 5, 3, 7, 8, 2]
#p = [1, 8, 2, 0, 3, 5, 6, 4, 7]

#p =  [1, 3, 2, 0, 5, 7, 6, 8, 4]
print("++++++++++A*++++++++++++")
astar(p)

问题分析与修复方案 #

原代码存在几个关键问题，直接导致了重复节点扩展：

1. 错误的节点扩展逻辑：每次循环都遍历所有已扩展节点生成子节点，这会重复生成旧节点的子节点，正确做法是只扩展当前选中的最小代价节点。
2. 未正确使用已访问集合：visit集合被定义但未实际使用，无法过滤已处理过的状态。
3. 代价计算错误：A星的总代价应为f(n) = g(n) + h(n)（g(n)是初始到当前的实际步数，h(n)是曼哈顿距离），原代码仅用h(n)作为代价，既无法保证最优路径，也容易引发重复节点。
4. 缺少重复节点过滤：生成子节点时未检查该状态是否已被处理或待处理，导致重复节点进入Fringe。

修改后的代码 #

import time

class Node:
    def __init__(self, state, g_cost, parent=None):
        self.parent = parent
        self.state = state
        self.g_cost = g_cost  # 初始节点到当前节点的实际步数
        self.h_cost = manhattan(state, [0,1,2,3,4,5,6,7,8])  # 曼哈顿距离启发值
        self.f_cost = self.g_cost + self.h_cost  # 总代价

def manhattan(a, b):
    # 修正曼哈顿距离计算：根据每个数字的目标坐标计算
    total = 0
    side = int(len(a)**0.5)
    for idx, val in enumerate(a):
        if val == 0:
            continue
        target_idx = b.index(val)
        # 计算当前坐标与目标坐标的曼哈顿距离
        total += abs((idx // side) - (target_idx // side)) + abs((idx % side) - (target_idx % side))
    return total

def astar(puzzle):
    start_time = time.time_ns()
    goal = [0,1,2,3,4,5,6,7,8]
    possible_moves = [[1,3],[0,2,4],[1,5],[0,4,6],[1,3,5,7],[2,4,8],[3,7],[4,6,8],[5,7]]
    
    # 初始化Fringe和已访问集合
    fringe = [Node(puzzle, 0)]
    visited = set()
    visited.add(tuple(puzzle))
    
    while fringe:
        # 找到Fringe中f_cost最小的节点
        current_node = min(fringe, key=lambda x: x.f_cost)
        fringe.remove(current_node)
        
        # 检查是否到达目标
        if current_node.state == goal:
            # 回溯获取路径
            path = []
            moves = 0
            while current_node.parent:
                path.append(current_node.state.index(8))
                current_node = current_node.parent
                moves += 1
            path.reverse()
            print(f"移动步数：{moves}")
            print(f"移动路径：{path}")
            end_time = time.time_ns()
            print(f"执行时间（纳秒）：{end_time - start_time}")
            return path
        
        # 标记当前节点为已访问
        visited.add(tuple(current_node.state))
        
        # 生成当前节点的所有子节点
        blank_idx = current_node.state.index(8)
        for move_idx in possible_moves[blank_idx]:
            # 生成新状态
            new_state = current_node.state.copy()
            new_state[blank_idx], new_state[move_idx] = new_state[move_idx], new_state[blank_idx]
            new_state_tuple = tuple(new_state)
            
            # 如果该状态已访问，跳过
            if new_state_tuple in visited:
                continue
            
            # 检查Fringe中是否已有该状态且代价更低
            exists_in_fringe = False
            for node in fringe:
                if tuple(node.state) == new_state_tuple:
                    exists_in_fringe = True
                    # 如果新路径的g_cost更低，更新该节点的代价和父节点
                    if current_node.g_cost + 1 < node.g_cost:
                        node.g_cost = current_node.g_cost + 1
                        node.f_cost = node.g_cost + node.h_cost
                        node.parent = current_node
                    break
            # 如果不在Fringe中，添加新节点
            if not exists_in_fringe:
                fringe.append(Node(new_state, current_node.g_cost + 1, current_node))
    
    # 无解的情况
    print("该状态无解")
    return None

# 测试用例
# p = [0, 1, 2, 3, 4, 5, 6, 8, 7]
# p = [0, 1, 2, 3, 8, 4, 6, 7, 5]
p = [6, 4, 0, 1, 5, 3, 7, 8, 2]
# p = [1, 8, 2, 0, 3, 5, 6, 4, 7]

print("++++++++++A*算法求解八数码++++++++++++")
astar(p)

关键修复点说明 #

• 正确的代价计算：拆分g_cost（实际步数）和h_cost（曼哈顿距离），总代价f_cost为两者之和，符合A星算法要求。
• 已访问集合过滤：用tuple存储状态（列表不可哈希），加入visited集合后，生成子节点时先检查是否已访问，直接跳过重复状态。
• 仅扩展当前最小代价节点：每次循环只处理Fringe中f_cost最小的节点，避免重复扩展旧节点。
• Fringe内节点优化：如果Fringe中已有相同状态但代价更高的节点，更新其代价和父节点，保证Fringe中都是当前最优的路径节点。

内容的提问来源于stack exchange，提问作者jwolf