从理论到实践:用Python实现《人工智能导论》第三章算法(王万良第五版)

很多朋友在学《人工智能导论》这类教材时,都有过类似的困惑:书上的公式、流程图和伪代码看着都懂,但总觉得隔着一层纱,离“真正理解”还差那么一口气。尤其是王万良老师这本经典教材的第三章,内容涵盖了搜索、推理等核心算法的理论基础,如果只停留在纸面推导,很容易陷入“一看就会,一写就废”的尴尬境地。我自己刚开始学的时候也这样,直到后来下定决心,把书里的每一个主要算法都用代码敲一遍,运行出来,看着控制台输出的结果一步步逼近答案,那种抽象概念瞬间落地的通透感,才让我真正觉得自己“会了”。

这篇文章,就是为你准备的“通关秘籍”。我们不谈空泛的理论,而是聚焦于动手实现。我会假设你已经具备了基础的Python编程能力,并且对教材第三章的内容有初步的阅读。我们的目标非常明确:将教材中的关键算法,从纸上的伪代码,变成你电脑里可以运行、可以调试、可以观察每一步变化的Python程序。通过这个过程,你不仅能验证理论,更能深刻理解算法背后的设计思想、效率考量以及那些在纯理论描述中容易被忽略的细节。这不仅仅是学习,更是一种“解构”与“重建”的思维训练。

1. 环境准备与基础框架搭建

在开始编码实现具体算法之前,花点时间搭建一个清晰、可复用的基础环境至关重要。这能让你后续的代码更加模块化,调试起来也更方便。我们不需要复杂的框架,几个核心的Python内置库就足够了。

首先,确保你的Python环境是3.7或以上版本。我们将主要用到 typing 模块来增强代码的可读性和类型提示,这对于实现复杂的数据结构(如图、树)非常有帮助。此外,collections 模块中的 deque(双端队列)在实现广度优先搜索(BFS)时会非常高效。

提示:虽然类型提示(Type Hints)不是强制性的,但它能极大地提升代码的清晰度和可维护性,尤其是在定义算法输入输出时,强烈建议养成使用的习惯。

为了给后续的搜索算法提供一个统一的“舞台”,我们先定义一个简单的“问题”抽象基类。这个类定义了所有搜索算法需要面对的问题的通用接口。

from abc import ABC, abstractmethod
from typing import List, Any, Optional, Tuple

class Problem(ABC):
    """问题抽象基类,定义搜索问题的通用接口。"""

    @abstractmethod
    def initial_state(self) -> Any:
        """返回问题的初始状态。"""
        pass

    @abstractmethod
    def is_goal(self, state: Any) -> bool:
        """判断给定状态是否为目标状态。"""
        pass

    @abstractmethod
    def actions(self, state: Any) -> List[Any]:
        """在给定状态下,返回所有可执行的动作列表。"""
        pass

    @abstractmethod
    def result(self, state: Any, action: Any) -> Any:
        """对给定状态执行一个动作,返回新的状态。"""
        pass

    @abstractmethod
    def step_cost(self, state: Any, action: Any, new_state: Any) -> float:
        """计算从状态state通过action到达new_state的单步代价。"""
        pass

这个 Problem 类就像一个契约,任何具体的问题(如八数码、路径规划)都需要继承它并实现这些抽象方法。这样做的好处是,我们之后编写的搜索算法(如BFS、DFS、A*)可以完全与具体问题解耦,它们只依赖于这个通用的接口。这是面向对象设计和算法泛化的重要实践。

接下来,我们还需要一个通用的“节点”类,用于在搜索树中记录状态、父节点、动作和累积代价等信息。

class Node:
    """搜索树中的节点。"""

    def __init__(self, state, parent=None, action=None, path_cost=0):
        self.state = state          # 该节点代表的状态
        self.parent = parent        # 父节点
        self.action = action        # 导致到达此状态的动作
        self.path_cost = path_cost  # 从初始状态到达此节点的路径总代价

    def __repr__(self):
        return f"Node(state={self.state}, cost={self.path_cost})"

    def __lt__(self, other):
        # 用于优先队列比较,这里默认按路径代价比较
        return self.path_cost < other.path_cost

有了 ProblemNode 这两个基础构件,我们就可以开始实现教材中的第一个重量级算法了。

2. 盲目搜索算法的Python实现:DFS与BFS

盲目搜索,也称为无信息搜索,是人工智能中最基础的搜索策略。它不利用问题领域的任何特定知识,只根据预定义的顺序(如深度优先、广度优先)系统地探索状态空间。王万良教材第三章对此有详细阐述,我们现在用代码将其具象化。

2.1 深度优先搜索(DFS)的实现与陷阱

深度优先搜索的策略是“一条道走到黑”,尽可能深地探索一条分支,直到无法继续(遇到死胡同或目标),然后回溯到上一个分叉点。其Python实现的核心在于使用栈(Stack) 这种后进先出(LIFO)的数据结构。

def depth_first_search(problem: Problem, depth_limit: int = None):
    """深度优先搜索算法。
    Args:
        problem: 问题实例,需实现Problem接口。
        depth_limit: 深度限制,用于防止无限递归,实现迭代加深搜索。
    Returns:
        如果找到目标,返回一个解序列(动作列表);否则返回None。
    """
    from collections import deque

    # 使用栈来管理待探索的节点边界
    frontier = deque([Node(problem.initial_state())])
    explored = set()  # 已探索状态集合,避免重复访问

    while frontier:
        node = frontier.pop()  # 栈:后进先出,弹出最后一个元素
        state = node.state

        if problem.is_goal(state):
            # 找到目标,回溯构建解路径
            return _build_solution(node)

        explored.add(state)

        # 如果设置了深度限制,且当前节点深度超过限制,则跳过扩展
        if depth_limit is not None and _node_depth(node) >= depth_limit:
            continue

        for action in problem.actions(state):
            child_state = problem.result(state, action)
            if child_state not in explored and not _in_frontier(frontier, child_state):
                child_node = Node(state=child_state, parent=node, action=action,
                                  path_cost=node.path_cost + problem.step_cost(state, action, child_state))
                frontier.append(child_node)  # 新节点压入栈顶
    return None  # 搜索失败

def _build_solution(node: Node) -> List:
    """从目标节点回溯到根节点,构建动作序列。"""
    actions = []
    while node.parent is not None:
        actions.append(node.action)
        node = node.parent
    actions.reverse()
    return actions

def _node_depth(node: Node) -> int:
    """计算节点的深度。"""
    depth = 0
    while node.parent is not None:
        depth += 1
        node = node.parent
    return depth

def _in_frontier(frontier, state):
    """检查某个状态是否已在边界集合中。"""
    for n in frontier:
        if n.state == state:
            return True
    return False

深度优先搜索的典型问题与改进: DFS最大的风险在于可能陷入无限深度的分支中(例如,状态图中有环),导致程序无法终止。上述代码通过 explored 集合记录了所有访问过的状态,有效避免了在同一路径上的循环。但即便如此,在状态空间无限或极深的情况下,DFS仍可能因为执着于一条错误路径而效率极低,甚至耗尽内存。这时就需要引入 深度限制(depth_limit),演变为迭代加深搜索(IDS)。IDS结合了DFS的空间效率和BFS的完备性优势,其实现就是在循环中不断增加深度限制,反复调用受限的DFS。

2.2 广度优先搜索(BFS)的实现与完备性

与DFS相反,广度优先搜索的策略是“层层推进”。它会先探索所有深度为1的节点,然后是深度为2的节点,以此类推。这种策略由队列(Queue) 这种先进先出(FIFO)的数据结构自然实现。

def breadth_first_search(problem: Problem):
    """广度优先搜索算法。
    Args:
        problem: 问题实例。
    Returns:
        如果找到目标,返回一个解序列(动作列表);否则返回None。
    """
    from collections import deque

    # 使用队列来管理待探索的节点边界
    frontier = deque([Node(problem.initial_state())])
    explored = set()

    while frontier:
        node = frontier.popleft()  # 队列:先进先出,弹出第一个元素
        state = node.state

        if problem.is_goal(state):
            return _build_solution(node)

        explored.add(state)

        for action in problem.actions(state):
            child_state = problem.result(state, action)
            if child_state not in explored and not _in_frontier(frontier, child_state):
                child_node = Node(state=child_state, parent=node, action=action,
                                  path_cost=node.path_cost + problem.step_cost(state, action, child_state))
                frontier.append(child_node)  # 新节点加入队尾
    return None

BFS的特性分析: BFS最显著的优点是完备性:只要问题有解,且分支因子有限,BFS就一定能找到解。并且,在动作代价一致的情况下,BFS找到的解一定是最短路径(步数最少)。然而,它的代价是巨大的空间开销。因为BFS需要存储每一层的所有节点,其空间复杂度是 O(b^d),其中b是分支因子,d是目标深度。对于深度较大的问题,这可能是灾难性的。下面的表格对比了DFS和BFS的核心特性:

特性维度深度优先搜索 (DFS)广度优先搜索 (BFS)
数据结构栈 (Stack)队列 (Queue)
空间复杂度O(bm),m为最大深度,通常远小于d,空间效率高O(b^d),需要存储所有边界节点,空间消耗大
是否完备在有限状态空间且避免重复访问时完备;无限空间下不完备当分支因子有限时完备
是否最优否(除非运气好)在单步代价一致时最优(找到最短路径)
适用场景解可能很深,或空间受限;常用于拓扑排序、连通分量检测等要求找到最短路径,且状态空间深度不大;常用于最短路径问题

盲目搜索是理解更高级算法的基础。通过亲手实现它们,你会对“状态空间”、“边界集”、“已探索集”这些概念有肌肉记忆般的理解。接下来,我们将引入问题的领域知识,让搜索变得“聪明”起来。

3. 启发式搜索的核心:A*算法实战

当搜索问题变得复杂,状态空间呈指数级增长时,盲目搜索就像在漆黑的迷宫里乱撞。启发式搜索为我们点亮了一盏“探照灯”——启发函数(Heuristic Function)。它利用问题领域的特定知识,估算从当前状态到目标状态的代价,从而引导搜索朝着最有希望的方向进行。A*算法是启发式搜索中最著名且最有效的算法之一,它巧妙地结合了实际已花费代价g(n)估计剩余代价h(n)

3.1 A*算法的原理与Python实现

A算法的核心思想是评价函数 f(n) = g(n) + h(n)。它总是优先扩展f(n)值最小的节点,其中g(n)是从起点到节点n的实际路径代价,h(n)是从节点n到目标节点的估计代价(启发函数)。A算法的最优性依赖于启发函数h(n)的可采纳性(Admissible),即h(n)永远不会高估到达目标的实际代价。

下面是一个使用优先队列(heapq模块)实现的A*算法:

import heapq

def a_star_search(problem: Problem, heuristic):
    """A*搜索算法。
    Args:
        problem: 问题实例。
        heuristic: 启发函数,接受一个状态,返回估计的剩余代价。
    Returns:
        如果找到目标,返回一个解序列(动作列表);否则返回None。
    """
    initial_node = Node(problem.initial_state())
    # 优先队列中的元素为 (f(n), 唯一标识, node)
    # 使用唯一标识(如id(node))是为了避免比较Node对象本身
    frontier = []
    heapq.heappush(frontier, (0 + heuristic(initial_node.state), id(initial_node), initial_node))
    frontier_state_dict = {initial_node.state: initial_node}  # 用于快速查找状态是否在边界中
    explored = set()
    g_score = {initial_node.state: 0}  # 记录到达每个状态的最佳g(n)值

    while frontier:
        _, _, node = heapq.heappop(frontier)
        state = node.state

        if problem.is_goal(state):
            return _build_solution(node)

        explored.add(state)
        # 从边界字典中移除
        if state in frontier_state_dict:
            del frontier_state_dict[state]

        for action in problem.actions(state):
            child_state = problem.result(state, action)
            tentative_g = g_score[state] + problem.step_cost(state, action, child_state)

            # 如果子节点已在已探索集,且新路径不比已知的好,则跳过
            if child_state in explored and tentative_g >= g_score.get(child_state, float('inf')):
                continue

            # 如果子节点不在边界,或者找到了到达该状态更优的路径
            if child_state not in frontier_state_dict or tentative_g < g_score.get(child_state, float('inf')):
                g_score[child_state] = tentative_g
                f_score = tentative_g + heuristic(child_state)
                child_node = Node(state=child_state, parent=node, action=action, path_cost=tentative_g)
                heapq.heappush(frontier, (f_score, id(child_node), child_node))
                frontier_state_dict[child_state] = child_node
                # 注意:如果节点已在边界且找到了更优路径,旧节点仍会在堆中,但会被新节点(更小的f值)覆盖,
                # 当旧节点被弹出时,其状态已在explored中,会被跳过。这是一种“惰性删除”策略。
    return None

这段代码有几个关键点:

  1. 优先队列的使用heapq 模块提供了基于二叉堆的最小优先队列,确保每次弹出的都是f(n)值最小的节点。
  2. g_score字典:它记录了到达每个状态的最佳(最小)实际代价。这是A*算法能够找到最优解的关键,因为它可以比较到达同一状态的不同路径,并保留代价更小的那条。
  3. 惰性删除:当我们在边界中发现到达某个状态的更优路径时,我们并没有从堆中物理删除旧的、较差的节点,而是直接插入新的、更优的节点。当旧的较差节点后来被弹出时,我们通过检查其g_score是否与当前记录的最佳g_score一致来判断它是否已被更新,如果不一致,则直接丢弃它。这种实现更简单高效。

3.2 启发函数的设计:以八数码问题为例

理论说了这么多,我们用一个经典的八数码问题(8-puzzle) 来具体感受一下。八数码问题在一个3x3的网格中摆放了8个编号方块和一个空格,目标是通过滑动方块,使网格呈现目标布局。

首先,我们定义八数码问题的具体类,实现 Problem 接口:

class EightPuzzleProblem(Problem):
    """八数码问题。"""

    def __init__(self, initial, goal=(1, 2, 3, 4, 5, 6, 7, 8, 0)):
        self.initial = initial
        self.goal = goal

    def initial_state(self):
        return self.initial

    def is_goal(self, state):
        return state == self.goal

    def actions(self, state):
        """返回可以移动的方向:'up', 'down', 'left', 'right'。"""
        actions = []
        idx = state.index(0)  # 空格的位置
        row, col = divmod(idx, 3)
        if row > 0:
            actions.append('up')
        if row < 2:
            actions.append('down')
        if col > 0:
            actions.append('left')
        if col < 2:
            actions.append('right')
        return actions

    def result(self, state, action):
        """执行移动,返回新状态(元组)。"""
        idx = state.index(0)
        new_list = list(state)
        row, col = divmod(idx, 3)
        if action == 'up':
            swap_idx = idx - 3
        elif action == 'down':
            swap_idx = idx + 3
        elif action == 'left':
            swap_idx = idx - 1
        elif action == 'right':
            swap_idx = idx + 1
        else:
            raise ValueError(f"Invalid action: {action}")
        # 交换空格和目标位置
        new_list[idx], new_list[swap_idx] = new_list[swap_idx], new_list[idx]
        return tuple(new_list)

    def step_cost(self, state, action, new_state):
        # 八数码问题通常假设每次移动代价为1
        return 1

现在,我们来设计两个经典的启发函数:

  1. 错位数(Misplaced Tiles):计算当前状态与目标状态相比,位置不正确的方块数量(不包括空格)。这个函数是可采纳的,因为恢复每个错位的方块至少需要一步。
  2. 曼哈顿距离(Manhattan Distance):计算每个方块当前位置到其目标位置的曼哈顿距离(水平距离+垂直距离)之和。这个函数也是可采纳的,并且通常比错位数更“知情”(Informed),能提供更好的搜索引导。
def misplaced_tiles_heuristic(state, goal=(1,2,3,4,5,6,7,8,0)):
    """错位数启发函数。"""
    return sum(1 for s, g in zip(state, goal) if s != g and s != 0)  # 空格不算

def manhattan_distance_heuristic(state, goal=(1,2,3,4,5,6,7,8,0)):
    """曼哈顿距离启发函数。"""
    distance = 0
    # 构建一个从数字到其在目标状态中坐标的映射
    goal_pos = {goal[i]: (i // 3, i % 3) for i in range(9)}
    for idx, num in enumerate(state):
        if num == 0:
            continue
        current_row, current_col = divmod(idx, 3)
        goal_row, goal_col = goal_pos[num]
        distance += abs(current_row - goal_row) + abs(current_col - goal_col)
    return distance

让我们用一个简单的例子来测试和对比:

# 定义一个可解的初始状态
initial_state = (1, 2, 3, 4, 0, 5, 7, 8, 6)  # 距离目标很近
problem = EightPuzzleProblem(initial_state)

print("使用错位数启发函数的A*搜索:")
solution_misplaced = a_star_search(problem, lambda s: misplaced_tiles_heuristic(s))
print(f"解序列: {solution_misplaced}")
print(f"步数: {len(solution_misplaced) if solution_misplaced else '无解'}")

print("\n使用曼哈顿距离启发函数的A*搜索:")
solution_manhattan = a_star_search(problem, lambda s: manhattan_distance_heuristic(s))
print(f"解序列: {solution_manhattan}")
print(f"步数: {len(solution_manhattan) if solution_manhattan else '无解'}")

运行这段代码,你会发现对于这个简单状态,两种启发函数都能快速找到解。但如果你尝试一个更复杂的初始状态(如 (8,7,6,5,4,3,2,1,0)),并统计算法扩展的节点数量,你会直观地看到曼哈顿距离启发函数的强大之处——它通常能引导算法探索更少的节点,更快地找到最优解。这正是启发函数“信息量”的体现:一个更准确(但仍然是可采纳的)的启发函数能显著提升搜索效率。

4. 对抗搜索与博弈树:Minimax算法解析

第三章另一个精彩的部分是对抗搜索,它模拟了两个或多个对手之间的竞争环境。最经典的模型是零和博弈,一方的收益意味着另一方的损失。Minimax算法是解决此类问题的理论基础,它假设对手是理性的,总是做出对自己最有利(对己方最不利)的决策。

4.1 Minimax算法的递归实现

想象一下井字棋(Tic-Tac-Toe)。轮到“我方”(MAX玩家)走棋时,我们希望选择能带来最高评估分数的走法;而轮到“对方”(MIN玩家)时,他们会选择能给我方带来最低分数的走法。Minimax算法通过递归地模拟这棵博弈树,直到终止状态(赢、输、平局)或达到深度限制,来为当前局面选择最优行动。

首先,我们需要一个博弈问题的通用接口:

class Game(ABC):
    """双人零和博弈抽象基类。"""

    @abstractmethod
    def initial_state(self):
        """返回初始游戏状态。"""
        pass

    @abstractmethod
    def player(self, state):
        """返回在当前状态下该行动的玩家('MAX' 或 'MIN')。"""
        pass

    @abstractmethod
    def actions(self, state):
        """返回在当前状态下所有合法的行动列表。"""
        pass

    @abstractmethod
    def result(self, state, action):
        """执行行动,返回新的游戏状态。"""
        pass

    @abstractmethod
    def terminal_test(self, state):
        """检查状态是否为终止状态(游戏结束)。"""
        pass

    @abstractmethod
    def utility(self, state):
        """在终止状态下,返回对MAX玩家的效用值(如赢:1,输:-1,平:0)。"""
        pass

基于这个接口,Minimax算法的递归实现非常直观:

def minimax_decision(state, game: Game):
    """给定一个状态,返回MAX玩家应采取的最佳行动。"""
    player = game.player(state)
    assert player == 'MAX', "minimax_decision should be called for MAX player's turn"

    best_action = None
    best_value = -float('inf')

    for action in game.actions(state):
        # 模拟对手(MIN)也会最优应对
        value = min_value(game.result(state, action), game)
        if value > best_value:
            best_value = value
            best_action = action
    return best_action

def max_value(state, game: Game):
    """计算在state下,MAX玩家的最大可能效用值。"""
    if game.terminal_test(state):
        return game.utility(state)
    v = -float('inf')
    for action in game.actions(state):
        v = max(v, min_value(game.result(state, action), game))
    return v

def min_value(state, game: Game):
    """计算在state下,MIN玩家的最小可能效用值(即对MAX最不利)。"""
    if game.terminal_test(state):
        return game.utility(state)
    v = float('inf')
    for action in game.actions(state):
        v = min(v, max_value(game.result(state, action), game))
    return v

这个实现简洁地反映了Minimax的思想:MAX层取子节点最大值,MIN层取子节点最小值,交替递归。然而,它的计算量是巨大的,需要遍历整棵博弈树。对于像象棋、围棋这样分支因子巨大的游戏,这是不可行的。

4.2 Alpha-Beta剪枝:极大提升搜索效率

Alpha-Beta剪枝是对Minimax的革命性优化。它能在不改变最终结果的前提下,剪掉那些不可能影响最终决策的分支,从而大幅减少需要评估的节点数量。其核心思想是维护两个值:

  • α:MAX玩家在当前路径上能保证的最佳(最大)值。
  • β:MIN玩家在当前路径上能保证的最佳(最小)值,即对MAX最不利的值。

在搜索过程中,如果发现某个节点的评估值已经超出了当前玩家的可行范围(对MAX来说,子节点值≤α;对MIN来说,子节点值≥β),那么该节点的其余分支就没有继续探索的必要了,可以直接“剪掉”。

def alpha_beta_search(state, game: Game):
    """带Alpha-Beta剪枝的Minimax搜索,返回最佳行动。"""
    player = game.player(state)
    assert player == 'MAX'
    best_action, _ = max_value_ab(state, game, -float('inf'), float('inf'))
    return best_action

def max_value_ab(state, game, alpha, beta):
    if game.terminal_test(state):
        return None, game.utility(state)
    best_action = None
    v = -float('inf')
    for action in game.actions(state):
        _, child_value = min_value_ab(game.result(state, action), game, alpha, beta)
        if child_value > v:
            v = child_value
            best_action = action
        # Alpha剪枝:如果v已经大于等于beta,MIN父节点不会选择这个分支
        if v >= beta:
            return best_action, v
        alpha = max(alpha, v)  # 更新alpha值
    return best_action, v

def min_value_ab(state, game, alpha, beta):
    if game.terminal_test(state):
        return None, game.utility(state)
    best_action = None
    v = float('inf')
    for action in game.actions(state):
        _, child_value = max_value_ab(game.result(state, action), game, alpha, beta)
        if child_value < v:
            v = child_value
            best_action = action
        # Beta剪枝:如果v已经小于等于alpha,MAX父节点不会选择这个分支
        if v <= alpha:
            return best_action, v
        beta = min(beta, v)  # 更新beta值
    return best_action, v

为了直观展示Alpha-Beta剪枝的效果,我们可以用一个简单的估值函数和博弈树来模拟。假设一个简单的游戏,其状态可以用一个整数表示,MAX玩家行动时,可以选择将数字+1或+2,先达到或超过5的玩家获胜(效用值1),否则对手获胜(效用值-1)。虽然这个游戏很简单,但足以演示剪枝过程。

class SimpleGame(Game):
    """一个演示用的简单数字游戏。"""
    def __init__(self, start_num=0, target=5):
        self.start = start_num
        self.target = target

    def initial_state(self):
        return self.start

    def player(self, state):
        return 'MAX' if state % 2 == 0 else 'MIN'  # 简单交替

    def actions(self, state):
        return [1, 2]  # 每次可以加1或加2

    def result(self, state, action):
        return state + action

    def terminal_test(self, state):
        return state >= self.target

    def utility(self, state):
        # 当前玩家(刚达到target的玩家)是MAX则赢,否则输
        # 由于达到target时刚行动完,所以根据行动前的玩家判断
        # 简化逻辑:如果state是偶数(MAX刚行动完)达到target,则MAX赢
        return 1 if state % 2 == 0 else -1

# 测试
game = SimpleGame(start_num=0, target=5)
print("Minimax决策(无剪枝):")
action = minimax_decision(game.initial_state(), game)
print(f"最佳行动: {action}")

print("\nAlpha-Beta搜索决策:")
action_ab = alpha_beta_search(game.initial_state(), game)
print(f"最佳行动: {action_ab}")

在实际运行中,你可以通过添加计数器来比较两种算法遍历的节点数。对于更复杂的游戏,Alpha-Beta剪枝通常能将搜索深度提高一倍,这意味着在相同时间内,它能评估更深的局面,做出更优的决策。这正是像国际象棋、围棋AI能够战胜人类冠军的核心技术基础之一。理解并实现了Minimax和Alpha-Beta剪枝,你就掌握了经典博弈论AI的钥匙。

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐