MCTS算法深度解析

class MCTSNode:
    def __init__(self, state, parent=None):
        self.state = state        # 游戏状态
        self.parent = parent      # 父节点
        self.children = []        # 子节点列表
        self.visits = 0          # 访问次数
        self.wins = 0            # 胜利次数
        self.untried_actions = state.get_legal_actions()
    
    def uct_select_child(self):
        """使用 UCT 公式选择子节点"""
        return max(self.children, key=lambda c: c.uct_value())
    
    def uct_value(self, c=1.41):
        """UCT = 平均胜率 + 探索项"""
        if self.visits == 0:
            return float('inf')
        
        exploitation = self.wins / self.visits
        exploration = c * np.sqrt(np.log(self.parent.visits) / self.visits)
        
        return exploitation + exploration
    
    def expand(self):
        """扩展一个未尝试的动作"""
        action = self.untried_actions.pop()
        next_state = self.state.apply_action(action)
        child = MCTSNode(next_state, parent=self)
        self.children.append(child)
        return child
    
    def update(self, result):
        """回传结果"""
        self.visits += 1
        self.wins += result

def mcts_search(root_state, num_simulations):
    """MCTS 主循环"""
    root = MCTSNode(root_state)
    
    for _ in range(num_simulations):
        node = root
        
        # 1. 选择：向下选择直到叶子节点或有未扩展动作的节点
        while node.untried_actions == [] and node.children != []:
            node = node.uct_select_child()
        
        # 2. 扩展：如果有未尝试的动作，扩展一个子节点
        if node.untried_actions != []:
            node = node.expand()
        
        # 3. 模拟：从当前节点随机对局到终局
        state = node.state
        while not state.is_terminal():
            action = random.choice(state.get_legal_actions())
            state = state.apply_action(action)
        
        result = state.get_result()  # 1 for win, 0 for loss, 0.5 for draw
        
        # 4. 回传：更新路径上所有节点的统计信息
        while node is not None:
            node.update(result)
            result = 1 - result  # 对手的结果相反
            node = node.parent
    
    # 返回访问次数最多的动作
    return max(root.children, key=lambda c: c.visits).state.last_action

import numpy as np

class UCTNode:
    def __init__(self, state, parent=None, prior_prob=1.0):
        self.state = state
        self.parent = parent
        self.children = {}
        
        self.visits = 0
        self.value_sum = 0  # 累计价值
        self.prior = prior_prob  # 先验概率（AlphaZero 中使用）
        
    def is_fully_expanded(self):
        return len(self.children) == len(self.state.get_legal_actions())
    
    def select_child(self, c_puct=1.0):
        """使用 UCT 或 PUCT 选择最佳子节点"""
        def uct_score(child):
            # Q值（平均奖励）
            Q = child.value_sum / child.visits if child.visits > 0 else 0
            
            # U值（探索奖励）
            U = c_puct * child.prior * np.sqrt(self.visits) / (1 + child.visits)
            
            return Q + U
        
        return max(self.children.values(), key=uct_score)
    
    def expand(self, action, next_state, prior_prob):
        """扩展新节点"""
        if action not in self.children:
            self.children[action] = UCTNode(
                state=next_state,
                parent=self,
                prior_prob=prior_prob
            )
        return self.children[action]
    
    def update(self, value):
        """更新节点统计"""
        self.visits += 1
        self.value_sum += value
    
    def get_value(self):
        """获取平均价值"""
        return self.value_sum / self.visits if self.visits > 0 else 0

# UCT 参数调优示例
def tune_uct_parameter(game, c_values):
    """测试不同 c 值的性能"""
    results = {}
    
    for c in c_values:
        wins = 0
        num_games = 100
        
        for _ in range(num_games):
            state = game.initial_state()
            
            while not state.is_terminal():
                action = mcts_search(state, num_simulations=1000, c_puct=c)
                state = state.apply_action(action)
            
            if state.winner() == 1:
                wins += 1
        
        results[c] = wins / num_games
        print(f"c = {c:.2f}: 胜率 = {results[c]:.2%}")
    
    best_c = max(results, key=results.get)
    return best_c, results

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

class AlphaZeroMCTS:
    def __init__(self, game, neural_network, num_simulations=800, c_puct=1.0):
        self.game = game
        self.network = neural_network
        self.num_simulations = num_simulations
        self.c_puct = c_puct
        self.root = None
    
    def search(self, state):
        """执行 MCTS 搜索"""
        if self.root is None or self.root.state != state:
            self.root = UCTNode(state)
        
        for _ in range(self.num_simulations):
            node = self.root
            search_path = [node]
            
            # 1. 选择：沿着 PUCT 最大的路径向下
            while node.is_fully_expanded() and node.children:
                node = node.select_child(self.c_puct)
                search_path.append(node)
            
            # 2. 扩展 + 评估（使用神经网络）
            if not node.state.is_terminal():
                # 神经网络预测：策略 P(s, a) 和价值 V(s)
                policy_probs, value = self.network.predict(node.state)
                
                # 扩展所有合法动作
                for action in node.state.get_legal_actions():
                    if action not in node.children:
                        next_state = node.state.apply_action(action)
                        prior_prob = policy_probs[action]
                        node.expand(action, next_state, prior_prob)
            else:
                # 终局状态
                value = node.state.get_result()
            
            # 3 & 4. 回传（隐式包含在 update 中）
            for n in reversed(search_path):
                n.update(value)
                value = -value  # 对手的视角
        
        return self.root
    
    def get_action_probs(self, state, temperature=1.0):
        """获取动作概率分布"""
        root = self.search(state)
        
        action_visits = [
            (action, child.visits)
            for action, child in root.children.items()
        ]
        
        actions, visits = zip(*action_visits)
        visits = np.array(visits)
        
        if temperature == 0:
            # 确定性：选择访问最多的
            probs = np.zeros(len(visits))
            probs[np.argmax(visits)] = 1.0
        else:
            # 按温度缩放后归一化
            visits = visits ** (1.0 / temperature)
            probs = visits / visits.sum()
        
        return actions, probs
    
    def get_best_action(self, state):
        """获取最佳动作"""
        actions, probs = self.get_action_probs(state, temperature=0)
        return actions[np.argmax(probs)]

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

# 创建 MCTS 实例
network = SimpleNeuralNetwork()
mcts = AlphaZeroMCTS(
    game=my_game,
    neural_network=network,
    num_simulations=800,
    c_puct=1.0
)

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

class RAVENode(UCTNode):
    def __init__(self, state, parent=None):
        super().__init__(state, parent)
        self.amaf_visits = {}  # {action: visit_count}
        self.amaf_value = {}   # {action: total_value}
    
    def update_amaf(self, action, value):
        """更新 AMAF 统计"""
        if action not in self.amaf_visits:
            self.amaf_visits[action] = 0
            self.amaf_value[action] = 0.0
        
        self.amaf_visits[action] += 1
        self.amaf_value[action] += value
    
    def rave_value(self, action, c=1.0, k=100):
        """计算 RAVE-UCT 值"""
        # 树统计
        child = self.children.get(action)
        if child is None:
            return float('inf')
        
        n_tree = child.visits
        if n_tree == 0:
            return float('inf')
        
        q_tree = child.value_sum / n_tree
        
        # AMAF 统计
        n_amaf = self.amaf_visits.get(action, 0)
        q_amaf = (self.amaf_value.get(action, 0) / n_amaf) if n_amaf > 0 else 0
        
        # 计算权重 β
        beta = n_amaf / (n_tree + n_amaf + 4 * k * k * n_tree * n_amaf)
        
        # 混合值
        q_mix = (1 - beta) * q_tree + beta * q_amaf
        
        # 加上探索项
        exploration = c * np.sqrt(np.log(self.visits) / n_tree)
        
        return q_mix + exploration
    
    def select_child_rave(self, c=1.0):
        """使用 RAVE 选择子节点"""
        return max(
            self.children.keys(),
            key=lambda a: self.rave_value(a, c)
        )

def mcts_with_rave(root_state, num_simulations):
    """带 RAVE 的 MCTS"""
    root = RAVENode(root_state)
    
    for _ in range(num_simulations):
        node = root
        path = [node]
        actions_in_simulation = []
        
        # 选择
        while node.is_fully_expanded() and node.children:
            action = node.select_child_rave()
            actions_in_simulation.append(action)
            node = node.children[action]
            path.append(node)
        
        # 扩展
        if not node.is_fully_expanded():
            action = random.choice(node.state.get_legal_actions())
            actions_in_simulation.append(action)
            next_state = node.state.apply_action(action)
            node = node.expand(action, next_state, 1.0)
            path.append(node)
        
        # 模拟
        state = node.state
        while not state.is_terminal():
            action = random.choice(state.get_legal_actions())
            actions_in_simulation.append(action)
            state = state.apply_action(action)
        
        result = state.get_result()
        
        # 回传（包括 AMAF 更新）
        for i, n in enumerate(reversed(path)):
            n.update(result)
            
            # 更新 AMAF：路径后续所有动作
            for action in actions_in_simulation[i:]:
                n.update_amaf(action, result)
            
            result = 1 - result
    
    return max(root.children.keys(), key=lambda a: root.children[a].visits)

class VirtualLossNode(UCTNode):
    def __init__(self, state, parent=None):
        super().__init__(state, parent)
        self.virtual_loss = 0
    
    def get_value_with_virtual_loss(self):
        """考虑虚拟损失的价值"""
        if self.visits == 0:
            return 0
        return (self.value_sum - self.virtual_loss) / (self.visits + self.virtual_loss)
    
    def add_virtual_loss(self, loss=1):
        """添加虚拟损失"""
        self.virtual_loss += loss
    
    def remove_virtual_loss(self, loss=1):
        """移除虚拟损失"""
        self.virtual_loss -= loss

import threading

class ParallelMCTS:
    def __init__(self, game, num_threads=4):
        self.game = game
        self.num_threads = num_threads
        self.lock = threading.Lock()
    
    def search_thread(self, root, num_simulations_per_thread):
        """单线程搜索"""
        for _ in range(num_simulations_per_thread):
            node = root
            path = []
            
            # 选择（加锁）
            with self.lock:
                while node.is_fully_expanded() and node.children:
                    node = node.select_child()
                    node.add_virtual_loss()  # 添加虚拟损失
                    path.append(node)
            
            # 扩展、模拟（不需要锁）
            if not node.state.is_terminal():
                # ... 扩展和模拟逻辑
                value = simulate(node.state)
            else:
                value = node.state.get_result()
            
            # 回传（加锁）
            with self.lock:
                for n in reversed(path):
                    n.update(value)
                    n.remove_virtual_loss()  # 移除虚拟损失
                    value = 1 - value
    
    def search(self, state, total_simulations):
        """并行搜索"""
        root = VirtualLossNode(state)
        simulations_per_thread = total_simulations // self.num_threads
        
        threads = []
        for _ in range(self.num_threads):
            t = threading.Thread(
                target=self.search_thread,
                args=(root, simulations_per_thread)
            )
            threads.append(t)
            t.start()
        
        for t in threads:
            t.join()
        
        return max(root.children.keys(), key=lambda a: root.children[a].visits)

class ProgressiveWideningNode(UCTNode):
    def __init__(self, state, parent=None, k=10, alpha=0.5):
        super().__init__(state, parent)
        self.k = k  # 扩展参数
        self.alpha = alpha  # 扩展指数
    
    def should_expand(self):
        """判断是否应该扩展新节点"""
        allowed_children = self.k * (self.visits ** self.alpha)
        return len(self.children) < allowed_children
    
    def expand_progressive(self):
        """渐进式扩展"""
        if self.should_expand() and self.untried_actions:
            # 选择最有希望的动作（可以用启发式或神经网络）
            action = self.select_best_untried_action()
            self.untried_actions.remove(action)
            
            next_state = self.state.apply_action(action)
            child = ProgressiveWideningNode(next_state, parent=self)
            self.children[action] = child
            
            return child
        return None
    
    def select_best_untried_action(self):
        """选择最有希望的未尝试动作"""
        # 简化版本：随机选择
        # 实际应用中可以用启发式或神经网络评分
        return random.choice(self.untried_actions)

import numpy as np
import random
from copy import deepcopy

class TicTacToe:
    """井字棋游戏"""
    def __init__(self):
        self.board = np.zeros((3, 3), dtype=int)
        self.current_player = 1
    
    def get_legal_actions(self):
        """获取合法动作"""
        return [(i, j) for i in range(3) for j in range(3) if self.board[i, j] == 0]
    
    def apply_action(self, action):
        """应用动作，返回新状态"""
        new_state = deepcopy(self)
        i, j = action
        new_state.board[i, j] = new_state.current_player
        new_state.current_player = -new_state.current_player
        return new_state
    
    def is_terminal(self):
        """判断是否终局"""
        return self.get_winner() is not None or len(self.get_legal_actions()) == 0
    
    def get_winner(self):
        """获取胜者：1, -1, 或 None"""
        # 检查行
        for i in range(3):
            if abs(self.board[i, :].sum()) == 3:
                return self.board[i, 0]
        
        # 检查列
        for j in range(3):
            if abs(self.board[:, j].sum()) == 3:
                return self.board[0, j]
        
        # 检查对角线
        if abs(np.trace(self.board)) == 3:
            return self.board[0, 0]
        if abs(np.trace(np.fliplr(self.board))) == 3:
            return self.board[0, 2]
        
        return None
    
    def get_result(self):
        """获取当前玩家的结果（1=胜，0=负，0.5=平）"""
        winner = self.get_winner()
        if winner == self.current_player:
            return 1.0
        elif winner == -self.current_player:
            return 0.0
        else:
            return 0.5
    
    def display(self):
        """显示棋盘"""
        symbols = {0: '.', 1: 'X', -1: 'O'}
        for i in range(3):
            print(' '.join(symbols[self.board[i, j]] for j in range(3)))
        print()

class TicTacToeMCTSNode:
    """井字棋 MCTS 节点"""
    def __init__(self, state, parent=None):
        self.state = state
        self.parent = parent
        self.children = {}
        self.visits = 0
        self.value_sum = 0
        self.untried_actions = state.get_legal_actions()
    
    def is_fully_expanded(self):
        return len(self.untried_actions) == 0
    
    def select_child(self, c=1.41):
        """UCT 选择"""
        def uct_value(child):
            if child.visits == 0:
                return float('inf')
            exploitation = child.value_sum / child.visits
            exploration = c * np.sqrt(np.log(self.visits) / child.visits)
            return exploitation + exploration
        
        return max(self.children.values(), key=uct_value)
    
    def expand(self):
        """扩展一个子节点"""
        action = self.untried_actions.pop()
        next_state = self.state.apply_action(action)
        child = TicTacToeMCTSNode(next_state, parent=self)
        self.children[action] = child
        return child
    
    def simulate(self):
        """随机模拟到终局"""
        state = deepcopy(self.state)
        while not state.is_terminal():
            action = random.choice(state.get_legal_actions())
            state = state.apply_action(action)
        return state.get_result()
    
    def backpropagate(self, result):
        """回传结果"""
        self.visits += 1
        self.value_sum += result
        if self.parent:
            self.parent.backpropagate(1 - result)

def mcts_tictactoe(state, num_simulations=1000):
    """井字棋 MCTS 搜索"""
    root = TicTacToeMCTSNode(state)
    
    for _ in range(num_simulations):
        node = root
        
        # 1. 选择
        while node.is_fully_expanded() and node.children:
            node = node.select_child()
        
        # 2. 扩展
        if not node.is_fully_expanded():
            node = node.expand()
        
        # 3. 模拟
        result = node.simulate()
        
        # 4. 回传
        node.backpropagate(result)
    
    # 返回访问次数最多的动作
    best_action = max(
        root.children.items(),
        key=lambda item: item[1].visits
    )[0]
    
    return best_action

# 测试：MCTS vs 随机玩家
def play_game():
    """完整对局"""
    game = TicTacToe()
    
    while not game.is_terminal():
        game.display()
        
        if game.current_player == 1:
            # MCTS 玩家
            action = mcts_tictactoe(game, num_simulations=1000)
            print(f"MCTS 选择: {action}")
        else:
            # 随机玩家
            action = random.choice(game.get_legal_actions())
            print(f"随机选择: {action}")
        
        game = game.apply_action(action)
    
    game.display()
    winner = game.get_winner()
    if winner == 1:
        print("MCTS 获胜！")
    elif winner == -1:
        print("随机玩家获胜！")
    else:
        print("平局！")

# play_game()

def evaluate_strength(num_simulations_list, num_games=100):
    """评估不同模拟次数的棋力"""
    results = {}
    
    for num_sims in num_simulations_list:
        wins = 0
        
        for _ in range(num_games):
            game = TicTacToe()
            
            while not game.is_terminal():
                if game.current_player == 1:
                    # MCTS 玩家
                    action = mcts_tictactoe(game, num_simulations=num_sims)
                else:
                    # 对手：固定模拟次数（如100次）
                    action = mcts_tictactoe(game, num_simulations=100)
                
                game = game.apply_action(action)
            
            if game.get_winner() == 1:
                wins += 1
        
        winrate = wins / num_games
        results[num_sims] = winrate
        print(f"模拟次数={num_sims}: 胜率={winrate:.2%}")
    
    return results

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

import matplotlib.pyplot as plt

def visualize_uct_exploration():
    """可视化 UCT 的探索-利用平衡"""
    N = 100  # 父节点访问次数
    c_values = [0.5, 1.0, 1.41, 2.0]
    
    n_range = np.arange(1, N+1)
    
    plt.figure(figsize=(12, 6))
    
    for c in c_values:
        exploration_bonus = c * np.sqrt(np.log(N) / n_range)
        plt.plot(n_range, exploration_bonus, label=f'c={c}', linewidth=2)
    
    plt.xlabel('节点访问次数 n')
    plt.ylabel('探索奖励')
    plt.title('UCT 探索项随访问次数的变化')
    plt.legend()
    plt.grid(alpha=0.3)
    plt.show()

# visualize_uct_exploration()