如何将给定的BFS（广度优先搜索）伪代码转换为可运行的Python代码

原提供的伪代码存在部分逻辑疏漏（比如初始节点未入队、找到目标后未提前返回、集合操作语法错误等），以下是修正后的可运行实现：

1. 基础依赖定义 #

首先定义通用节点类，用来存储状态、父节点、路径成本等信息：

class Node:
    def __init__(self, state, parent=None, path_cost=0):
        self.state = state  # 当前节点状态
        self.parent = parent  # 父节点，用于回溯完整路径
        self.path_cost = path_cost  # 从初始节点到当前节点的路径成本

其次定义Problem抽象接口，你需要根据自己的业务场景实现对应方法：

class Problem:
    def __init__(self, initial_state, goal_state=None):
        self.INITIAL = initial_state  # 初始状态
        self.goal_state = goal_state  # 目标状态
    
    def IS_GOAL(self, state):
        """判断当前状态是否符合目标要求，可自定义判断逻辑"""
        return state == self.goal_state
    
    def EXPAND(self, node):
        """展开当前节点，返回所有合法子节点列表，必须根据业务逻辑实现"""
        children = []
        # 示例：如果是网格路径查找，可扩展上下左右四个方向的节点
        # x, y = node.state
        # for dx, dy in [(-1,0), (1,0), (0,-1), (0,1)]:
        #     new_state = (x+dx, y+dy)
        #     children.append(Node(new_state, parent=node, path_cost=node.path_cost+1))
        return children

2. BFS核心实现 #

from collections import deque

def BFS(problem):
    # 初始化初始节点
    initial_node = Node(problem.INITIAL)
    # 提前判断初始状态是否就是目标
    if problem.IS_GOAL(initial_node.state):
        return initial_node
    # 初始化前沿队列：BFS用FIFO队列保证按层级遍历
    frontier = deque([initial_node])
    # 已访问状态集合，避免重复遍历
    reached = {problem.INITIAL}
    
    while frontier:
        # 取出队列最前端的节点
        current_node = frontier.popleft()
        # 展开当前节点的所有子节点
        for child in problem.EXPAND(current_node):
            s = child.state
            # 找到目标直接返回结果
            if problem.IS_GOAL(s):
                return child
            # 未访问过的状态加入队列和已访问集合
            if s not in reached:
                reached.add(s)
                frontier.append(child)
    # 遍历所有节点未找到目标，返回None代表搜索失败
    return None

3. UCS（一致代价搜索）实现说明 #

你提供的伪代码中UCS逻辑有误，UCS需要按路径成本从小到大优先遍历，不能直接调用BFS，需要用优先队列替代普通FIFO队列，实现如下：

import heapq

def UCS(problem):
    initial_node = Node(problem.INITIAL)
    if problem.IS_GOAL(initial_node.state):
        return initial_node
    # 优先队列，存储格式为(路径成本, 节点)
    frontier = []
    heapq.heappush(frontier, (initial_node.path_cost, initial_node))
    reached = {problem.INITIAL: initial_node.path_cost}
    
    while frontier:
        current_cost, current_node = heapq.heappop(frontier)
        if problem.IS_GOAL(current_node.state):
            return current_node
        for child in problem.EXPAND(current_node):
            s = child.state
            # 只有新路径成本比已记录的更低时才更新
            if s not in reached or child.path_cost < reached[s]:
                reached[s] = child.path_cost
                heapq.heappush(frontier, (child.path_cost, child))
    return None

使用示例 #

以5*5网格从(0,0)到(3,3)的路径查找为例：

class GridProblem(Problem):
    def EXPAND(self, node):
        children = []
        x, y = node.state
        # 允许上下左右四个方向移动
        for dx, dy in [(-1,0), (1,0), (0,-1), (0,1)]:
            nx, ny = x + dx, y + dy
            # 限制网格范围
            if 0 <= nx <=5 and 0 <= ny <=5:
                children.append(Node((nx, ny), parent=node, path_cost=node.path_cost+1))
        return children

# 初始化问题并调用BFS
problem = GridProblem(initial_state=(0,0), goal_state=(3,3))
result_node = BFS(problem)

# 找到目标后回溯输出路径
if result_node:
    path = []
    node = result_node
    while node:
        path.append(node.state)
        node = node.parent
    path.reverse()
    print(f"找到路径，长度：{len(path)-1}，路径：{path}")
else:
    print("未找到可行路径")

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