MCTS算法深度解析

1
class MCTSNode:
2
    def __init__(self, state, parent=None):
3
        self.state = state        # 游戏状态
4
        self.parent = parent      # 父节点
5
        self.children = []        # 子节点列表
6
        self.visits = 0          # 访问次数
7
        self.wins = 0            # 胜利次数
8
        self.untried_actions = state.get_legal_actions()
9

10
    def uct_select_child(self):
11
        """使用 UCT 公式选择子节点"""
12
        return max(self.children, key=lambda c: c.uct_value())
13

14
    def uct_value(self, c=1.41):
15
        """UCT = 平均胜率 + 探索项"""
16
        if self.visits == 0:
17
            return float('inf')
18

19
        exploitation = self.wins / self.visits
20
        exploration = c * np.sqrt(np.log(self.parent.visits) / self.visits)
21

22
        return exploitation + exploration
23

24
    def expand(self):
25
        """扩展一个未尝试的动作"""
26
        action = self.untried_actions.pop()
27
        next_state = self.state.apply_action(action)
28
        child = MCTSNode(next_state, parent=self)
29
        self.children.append(child)
30
        return child
31

32
    def update(self, result):
33
        """回传结果"""
34
        self.visits += 1
35
        self.wins += result
36

37
def mcts_search(root_state, num_simulations):
38
    """MCTS 主循环"""
39
    root = MCTSNode(root_state)
40

41
    for _ in range(num_simulations):
42
        node = root
43

44
        # 1. 选择：向下选择直到叶子节点或有未扩展动作的节点
45
        while node.untried_actions == [] and node.children != []:
46
            node = node.uct_select_child()
47

48
        # 2. 扩展：如果有未尝试的动作，扩展一个子节点
49
        if node.untried_actions != []:
50
            node = node.expand()
51

52
        # 3. 模拟：从当前节点随机对局到终局
53
        state = node.state
54
        while not state.is_terminal():
55
            action = random.choice(state.get_legal_actions())
56
            state = state.apply_action(action)
57

58
        result = state.get_result()  # 1 for win, 0 for loss, 0.5 for draw
59

60
        # 4. 回传：更新路径上所有节点的统计信息
61
        while node is not None:
62
            node.update(result)
63
            result = 1 - result  # 对手的结果相反
64
            node = node.parent
65

66
    # 返回访问次数最多的动作
67
    return max(root.children, key=lambda c: c.visits).state.last_action

1
import numpy as np
2

3
class UCTNode:
4
    def __init__(self, state, parent=None, prior_prob=1.0):
5
        self.state = state
6
        self.parent = parent
7
        self.children = {}
8

9
        self.visits = 0
10
        self.value_sum = 0  # 累计价值
11
        self.prior = prior_prob  # 先验概率（AlphaZero 中使用）
12

13
    def is_fully_expanded(self):
14
        return len(self.children) == len(self.state.get_legal_actions())
15

16
    def select_child(self, c_puct=1.0):
17
        """使用 UCT 或 PUCT 选择最佳子节点"""
18
        def uct_score(child):
19
            # Q值（平均奖励）
20
            Q = child.value_sum / child.visits if child.visits > 0 else 0
21

22
            # U值（探索奖励）
23
            U = c_puct * child.prior * np.sqrt(self.visits) / (1 + child.visits)
24

25
            return Q + U
26

27
        return max(self.children.values(), key=uct_score)
28

29
    def expand(self, action, next_state, prior_prob):
30
        """扩展新节点"""
31
        if action not in self.children:
32
            self.children[action] = UCTNode(
33
                state=next_state,
34
                parent=self,
35
                prior_prob=prior_prob
36
            )
37
        return self.children[action]
38

39
    def update(self, value):
40
        """更新节点统计"""
41
        self.visits += 1
42
        self.value_sum += value
43

44
    def get_value(self):
45
        """获取平均价值"""
46
        return self.value_sum / self.visits if self.visits > 0 else 0
47

48
# UCT 参数调优示例
49
def tune_uct_parameter(game, c_values):
50
    """测试不同 c 值的性能"""
51
    results = {}
52

53
    for c in c_values:
54
        wins = 0
55
        num_games = 100
56

57
        for _ in range(num_games):
58
            state = game.initial_state()
59

60
            while not state.is_terminal():
61
                action = mcts_search(state, num_simulations=1000, c_puct=c)
62
                state = state.apply_action(action)
63

64
            if state.winner() == 1:
65
                wins += 1
66

67
        results[c] = wins / num_games
68
        print(f"c = {c:.2f}: 胜率 = {results[c]:.2%}")
69

70
    best_c = max(results, key=results.get)
71
    return best_c, results
72

73
# 示例：测试不同的 c 值
74
c_values = [0.5, 1.0, 1.41, 2.0, 3.0]
75
# best_c, results = tune_uct_parameter(my_game, c_values)

1
class AlphaZeroMCTS:
2
    def __init__(self, game, neural_network, num_simulations=800, c_puct=1.0):
3
        self.game = game
4
        self.network = neural_network
5
        self.num_simulations = num_simulations
6
        self.c_puct = c_puct
7
        self.root = None
8

9
    def search(self, state):
10
        """执行 MCTS 搜索"""
11
        if self.root is None or self.root.state != state:
12
            self.root = UCTNode(state)
13

14
        for _ in range(self.num_simulations):
15
            node = self.root
16
            search_path = [node]
17

18
            # 1. 选择：沿着 PUCT 最大的路径向下
19
            while node.is_fully_expanded() and node.children:
20
                node = node.select_child(self.c_puct)
21
                search_path.append(node)
22

23
            # 2. 扩展 + 评估（使用神经网络）
24
            if not node.state.is_terminal():
25
                # 神经网络预测：策略 P(s, a) 和价值 V(s)
26
                policy_probs, value = self.network.predict(node.state)
27

28
                # 扩展所有合法动作
29
                for action in node.state.get_legal_actions():
30
                    if action not in node.children:
31
                        next_state = node.state.apply_action(action)
32
                        prior_prob = policy_probs[action]
33
                        node.expand(action, next_state, prior_prob)
34
            else:
35
                # 终局状态
36
                value = node.state.get_result()
37

38
            # 3 & 4. 回传（隐式包含在 update 中）
39
            for n in reversed(search_path):
40
                n.update(value)
41
                value = -value  # 对手的视角
42

43
        return self.root
44

45
    def get_action_probs(self, state, temperature=1.0):
46
        """获取动作概率分布"""
47
        root = self.search(state)
48

49
        action_visits = [
50
            (action, child.visits)
51
            for action, child in root.children.items()
52
        ]
53

54
        actions, visits = zip(*action_visits)
55
        visits = np.array(visits)
56

57
        if temperature == 0:
58
            # 确定性：选择访问最多的
59
            probs = np.zeros(len(visits))
60
            probs[np.argmax(visits)] = 1.0
61
        else:
62
            # 按温度缩放后归一化
63
            visits = visits ** (1.0 / temperature)
64
            probs = visits / visits.sum()
65

66
        return actions, probs
67

68
    def get_best_action(self, state):
69
        """获取最佳动作"""
70
        actions, probs = self.get_action_probs(state, temperature=0)
71
        return actions[np.argmax(probs)]
72

73
# 使用示例
74
class SimpleNeuralNetwork:
75
    """简化的神经网络接口"""
76
    def predict(self, state):
77
        """
78
        返回：
79
        - policy: dict {action: probability}
80
        - value: float (当前玩家的胜率预测)
81
        """
82
        # 实际实现中，这里调用深度神经网络
83
        legal_actions = state.get_legal_actions()
84
        policy = {a: 1.0/len(legal_actions) for a in legal_actions}
85
        value = 0.0  # 简化：假设平局
86
        return policy, value
87

88
# 创建 MCTS 实例
89
network = SimpleNeuralNetwork()
90
mcts = AlphaZeroMCTS(
91
    game=my_game,
92
    neural_network=network,
93
    num_simulations=800,
94
    c_puct=1.0
95
)
96

97
# 搜索并选择动作
98
state = my_game.initial_state()
99
best_action = mcts.get_best_action(state)

1
class RAVENode(UCTNode):
2
    def __init__(self, state, parent=None):
3
        super().__init__(state, parent)
4
        self.amaf_visits = {}  # {action: visit_count}
5
        self.amaf_value = {}   # {action: total_value}
6

7
    def update_amaf(self, action, value):
8
        """更新 AMAF 统计"""
9
        if action not in self.amaf_visits:
10
            self.amaf_visits[action] = 0
11
            self.amaf_value[action] = 0.0
12

13
        self.amaf_visits[action] += 1
14
        self.amaf_value[action] += value
15

16
    def rave_value(self, action, c=1.0, k=100):
17
        """计算 RAVE-UCT 值"""
18
        # 树统计
19
        child = self.children.get(action)
20
        if child is None:
21
            return float('inf')
22

23
        n_tree = child.visits
24
        if n_tree == 0:
25
            return float('inf')
26

27
        q_tree = child.value_sum / n_tree
28

29
        # AMAF 统计
30
        n_amaf = self.amaf_visits.get(action, 0)
31
        q_amaf = (self.amaf_value.get(action, 0) / n_amaf) if n_amaf > 0 else 0
32

33
        # 计算权重 β
34
        beta = n_amaf / (n_tree + n_amaf + 4 * k * k * n_tree * n_amaf)
35

36
        # 混合值
37
        q_mix = (1 - beta) * q_tree + beta * q_amaf
38

39
        # 加上探索项
40
        exploration = c * np.sqrt(np.log(self.visits) / n_tree)
41

42
        return q_mix + exploration
43

44
    def select_child_rave(self, c=1.0):
45
        """使用 RAVE 选择子节点"""
46
        return max(
47
            self.children.keys(),
48
            key=lambda a: self.rave_value(a, c)
49
        )
50

51
def mcts_with_rave(root_state, num_simulations):
52
    """带 RAVE 的 MCTS"""
53
    root = RAVENode(root_state)
54

55
    for _ in range(num_simulations):
56
        node = root
57
        path = [node]
58
        actions_in_simulation = []
59

60
        # 选择
61
        while node.is_fully_expanded() and node.children:
62
            action = node.select_child_rave()
63
            actions_in_simulation.append(action)
64
            node = node.children[action]
65
            path.append(node)
66

67
        # 扩展
68
        if not node.is_fully_expanded():
69
            action = random.choice(node.state.get_legal_actions())
70
            actions_in_simulation.append(action)
71
            next_state = node.state.apply_action(action)
72
            node = node.expand(action, next_state, 1.0)
73
            path.append(node)
74

75
        # 模拟
76
        state = node.state
77
        while not state.is_terminal():
78
            action = random.choice(state.get_legal_actions())
79
            actions_in_simulation.append(action)
80
            state = state.apply_action(action)
81

82
        result = state.get_result()
83

84
        # 回传（包括 AMAF 更新）
85
        for i, n in enumerate(reversed(path)):
86
            n.update(result)
87

88
            # 更新 AMAF：路径后续所有动作
89
            for action in actions_in_simulation[i:]:
90
                n.update_amaf(action, result)
91

92
            result = 1 - result
93

94
    return max(root.children.keys(), key=lambda a: root.children[a].visits)

1
class VirtualLossNode(UCTNode):
2
    def __init__(self, state, parent=None):
3
        super().__init__(state, parent)
4
        self.virtual_loss = 0
5

6
    def get_value_with_virtual_loss(self):
7
        """考虑虚拟损失的价值"""
8
        if self.visits == 0:
9
            return 0
10
        return (self.value_sum - self.virtual_loss) / (self.visits + self.virtual_loss)
11

12
    def add_virtual_loss(self, loss=1):
13
        """添加虚拟损失"""
14
        self.virtual_loss += loss
15

16
    def remove_virtual_loss(self, loss=1):
17
        """移除虚拟损失"""
18
        self.virtual_loss -= loss
19

20
import threading
21

22
class ParallelMCTS:
23
    def __init__(self, game, num_threads=4):
24
        self.game = game
25
        self.num_threads = num_threads
26
        self.lock = threading.Lock()
27

28
    def search_thread(self, root, num_simulations_per_thread):
29
        """单线程搜索"""
30
        for _ in range(num_simulations_per_thread):
31
            node = root
32
            path = []
33

34
            # 选择（加锁）
35
            with self.lock:
36
                while node.is_fully_expanded() and node.children:
37
                    node = node.select_child()
38
                    node.add_virtual_loss()  # 添加虚拟损失
39
                    path.append(node)
40

41
            # 扩展、模拟（不需要锁）
42
            if not node.state.is_terminal():
43
                # ... 扩展和模拟逻辑
44
                value = simulate(node.state)
45
            else:
46
                value = node.state.get_result()
47

48
            # 回传（加锁）
49
            with self.lock:
50
                for n in reversed(path):
51
                    n.update(value)
52
                    n.remove_virtual_loss()  # 移除虚拟损失
53
                    value = 1 - value
54

55
    def search(self, state, total_simulations):
56
        """并行搜索"""
57
        root = VirtualLossNode(state)
58
        simulations_per_thread = total_simulations // self.num_threads
59

60
        threads = []
61
        for _ in range(self.num_threads):
62
            t = threading.Thread(
63
                target=self.search_thread,
64
                args=(root, simulations_per_thread)
65
            )
66
            threads.append(t)
67
            t.start()
68

69
        for t in threads:
70
            t.join()
71

72
        return max(root.children.keys(), key=lambda a: root.children[a].visits)

1
class ProgressiveWideningNode(UCTNode):
2
    def __init__(self, state, parent=None, k=10, alpha=0.5):
3
        super().__init__(state, parent)
4
        self.k = k  # 扩展参数
5
        self.alpha = alpha  # 扩展指数
6

7
    def should_expand(self):
8
        """判断是否应该扩展新节点"""
9
        allowed_children = self.k * (self.visits ** self.alpha)
10
        return len(self.children) < allowed_children
11

12
    def expand_progressive(self):
13
        """渐进式扩展"""
14
        if self.should_expand() and self.untried_actions:
15
            # 选择最有希望的动作（可以用启发式或神经网络）
16
            action = self.select_best_untried_action()
17
            self.untried_actions.remove(action)
18

19
            next_state = self.state.apply_action(action)
20
            child = ProgressiveWideningNode(next_state, parent=self)
21
            self.children[action] = child
22

23
            return child
24
        return None
25

26
    def select_best_untried_action(self):
27
        """选择最有希望的未尝试动作"""
28
        # 简化版本：随机选择
29
        # 实际应用中可以用启发式或神经网络评分
30
        return random.choice(self.untried_actions)

1
def evaluate_strength(num_simulations_list, num_games=100):
2
    """评估不同模拟次数的棋力"""
3
    results = {}
4

5
    for num_sims in num_simulations_list:
6
        wins = 0
7

8
        for _ in range(num_games):
9
            game = TicTacToe()
10

11
            while not game.is_terminal():
12
                if game.current_player == 1:
13
                    # MCTS 玩家
14
                    action = mcts_tictactoe(game, num_simulations=num_sims)
15
                else:
16
                    # 对手：固定模拟次数（如100次）
17
                    action = mcts_tictactoe(game, num_simulations=100)
18

19
                game = game.apply_action(action)
20

21
            if game.get_winner() == 1:
22
                wins += 1
23

24
        winrate = wins / num_games
25
        results[num_sims] = winrate
26
        print(f"模拟次数={num_sims}: 胜率={winrate:.2%}")
27

28
    return results
29

30
# 测试
31
# simulations = [10, 50, 100, 500, 1000]
32
# results = evaluate_strength(simulations, num_games=50)

1
import matplotlib.pyplot as plt
2

3
def visualize_uct_exploration():
4
    """可视化 UCT 的探索-利用平衡"""
5
    N = 100  # 父节点访问次数
6
    c_values = [0.5, 1.0, 1.41, 2.0]
7

8
    n_range = np.arange(1, N+1)
9

10
    plt.figure(figsize=(12, 6))
11

12
    for c in c_values:
13
        exploration_bonus = c * np.sqrt(np.log(N) / n_range)
14
        plt.plot(n_range, exploration_bonus, label=f'c={c}', linewidth=2)
15

16
    plt.xlabel('节点访问次数 n')
17
    plt.ylabel('探索奖励')
18
    plt.title('UCT 探索项随访问次数的变化')
19
    plt.legend()
20
    plt.grid(alpha=0.3)
21
    plt.show()
22

23
# visualize_uct_exploration()