Upload 3 files

2048的网页实现.py

@@ -0,0 +1,293 @@
+import gradio as gr
+import numpy as np
+import random
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from game2048 import Game2048
+
+# 创建游戏实例
+game = Game2048(size=4)
+
+# 方块颜色映射（根据数字值）
+TILE_COLORS = {
+    0: "#cdc1b4",     # 空白格子
+    2: "#eee4da",     # 2
+    4: "#ede0c8",     # 4
+    8: "#f2b179",     # 8
+    16: "#f59563",    # 16
+    32: "#f67c5f",    # 32
+    64: "#f65e3b",    # 64
+    128: "#edcf72",   # 128
+    256: "#edcc61",   # 256
+    512: "#edc850",   # 512
+    1024: "#edc53f",  # 1024
+    2048: "#edc22e",  # 2048
+    4096: "#3c3a32",  # 4096+
+}
+
+# 文本颜色映射（根据背景深浅）
+TEXT_COLORS = {
+    0: "#776e65",     # 空白格子
+    2: "#776e65",     # 2
+    4: "#776e65",     # 4
+    8: "#f9f6f2",     # 8+
+    16: "#f9f6f2",    # 16+
+    32: "#f9f6f2",    # 32+
+    64: "#f9f6f2",    # 64+
+    128: "#f9f6f2",   # 128+
+    256: "#f9f6f2",   # 256+
+    512: "#f9f6f2",   # 512+
+    1024: "#f9f6f2",  # 1024+
+    2048: "#f9f6f2",  # 2048+
+    4096: "#f9f6f2",  # 4096+
+}
+
+# 定义DQN网络结构（与训练时相同）
+class DQN(nn.Module):
+    def __init__(self, input_channels, output_size):
+        super(DQN, self).__init__()
+        self.input_channels = input_channels
+        
+        # 卷积层
+        self.conv1 = nn.Conv2d(input_channels, 128, kernel_size=3, padding=1)
+        self.conv2 = nn.Conv2d(128, 128, kernel_size=3, padding=1)
+        self.conv3 = nn.Conv2d(128, 128, kernel_size=3, padding=1)
+        
+        # Dueling DQN架构
+        # 价值流
+        self.value_conv = nn.Conv2d(128, 4, kernel_size=1)
+        self.value_fc1 = nn.Linear(4 * 4 * 4, 128)
+        self.value_fc2 = nn.Linear(128, 1)
+        
+        # 优势流
+        self.advantage_conv = nn.Conv2d(128, 16, kernel_size=1)
+        self.advantage_fc1 = nn.Linear(16 * 4 * 4, 128)
+        self.advantage_fc2 = nn.Linear(128, output_size)
+        
+    def forward(self, x):
+        x = F.relu(self.conv1(x))
+        x = F.relu(self.conv2(x))
+        x = F.relu(self.conv3(x))
+        
+        # 价值流
+        value = F.relu(self.value_conv(x))
+        value = value.view(value.size(0), -1)
+        value = F.relu(self.value_fc1(value))
+        value = self.value_fc2(value)
+        
+        # 优势流
+        advantage = F.relu(self.advantage_conv(x))
+        advantage = advantage.view(advantage.size(0), -1)
+        advantage = F.relu(self.advantage_fc1(advantage))
+        advantage = self.advantage_fc2(advantage)
+        
+        # 合并价值流和优势流
+        q_values = value + advantage - advantage.mean(dim=1, keepdim=True)
+        return q_values
+
+# 加载模型
+def load_model(model_path):
+    model = DQN(4, 4)  # 输入通道4，输出动作4个
+    try:
+        checkpoint = torch.load(model_path, map_location=torch.device('cpu'))
+        model.load_state_dict(checkpoint['policy_net_state_dict'])
+        model.eval()
+        print("模型加载成功")
+        return model
+    except Exception as e:
+        print(f"模型加载失败: {e}")
+        return None
+
+# 尝试加载模型
+model_path = "models/dqn_2048_best_tile.pth"
+model = load_model(model_path)
+
+def render_board(board):
+    html = "<div style='background-color:#bbada0; padding:10px; border-radius:6px;'>"
+    html += "<table style='border-spacing:10px; border-collapse:separate;'>"
+    
+    for i in range(game.size):
+        html += "<tr>"
+        for j in range(game.size):
+            value = board[i][j]
+            color = TILE_COLORS.get(value, "#3c3a32")  # 默认深色
+            text_color = TEXT_COLORS.get(value, "#f9f6f2")  # 默认浅色
+            font_size = "36px" if value < 100 else "30px" if value < 1000 else "24px"
+            
+            html += f"""
+            <td style='background-color:{color}; 
+                        width:80px; height:80px; 
+                        border-radius:4px; 
+                        text-align:center; 
+                        font-weight:bold; 
+                        font-size:{font_size};
+                        color:{text_color};'>
+                {value if value > 0 else ''}
+            </td>
+            """
+        html += "</tr>"
+    
+    html += "</table></div>"
+    return html
+
+def make_move(direction):
+    """执行移动操作并更新界面"""
+    direction_names = ["上", "右", "下", "左"]
+    
+    # 执行移动
+    new_board, game_over = game.move(direction)
+    
+    # 渲染棋盘
+    board_html = render_board(new_board)
+    
+    # 更新状态信息
+    status = f"<b>移动方向:</b> {direction_names[direction]}"
+    status += f"<br><b>当前分数:</b> {game.score}"
+    status += f"<br><b>最大方块:</b> {np.max(game.board)}"
+    
+    if game.game_over:
+        status += "<br><br><div style='color:#ff0000; font-weight:bold;'>游戏结束!</div>"
+        status += f"<br><b>最终分数:</b> {game.score}"
+    
+    return board_html, status
+
+def reset_game():
+    """重置游戏"""
+    global game
+    game = Game2048(size=4)
+    board = game.reset()
+    
+    # 渲染棋盘
+    board_html = render_board(board)
+    
+    # 初始状态信息
+    status = "<b>游戏已重置!</b>"
+    status += f"<br><b>当前分数:</b> {game.score}"
+    status += f"<br><b>最大方块:</b> {np.max(game.board)}"
+    
+    return board_html, status
+
+def ai_move():
+    """使用AI模型进行一步移动"""
+    if model is None:
+        return render_board(game.board), "<b>错误:</b> 未加载AI模型"
+    
+    # 获取当前状态
+    state = game.get_state()
+    
+    # 获取有效移动
+    valid_moves = game.get_valid_moves()
+    if not valid_moves:
+        return render_board(game.board), "<b>游戏结束!</b> 没有有效移动"
+    
+    # 转换状态为模型输入
+    state_tensor = torch.tensor(state, dtype=torch.float).unsqueeze(0)
+    
+    # 模型预测
+    with torch.no_grad():
+        q_values = model(state_tensor).numpy().flatten()
+    
+    # 只考虑有效动作
+    valid_q_values = np.full(4, -np.inf)
+    for move in valid_moves:
+        valid_q_values[move] = q_values[move]
+    
+    # 选择最佳动作
+    action = np.argmax(valid_q_values)
+    
+    # 执行移动
+    direction_names = ["上", "右", "下", "左"]
+    new_board, game_over = game.move(action)
+    
+    # 渲染棋盘
+    board_html = render_board(new_board)
+    
+    # 更新状态信息
+    status = f"<b>AI移动方向:</b> {direction_names[action]}"
+    status += f"<br><b>当前分数:</b> {game.score}"
+    status += f"<br><b>最大方块:</b> {np.max(game.board)}"
+    
+    if game.game_over:
+        status += "<br><br><div style='color:#ff0000; font-weight:bold;'>游戏结束!</div>"
+        status += f"<br><b>最终分数:</b> {game.score}"
+    
+    return board_html, status
+
+# 创建Gradio界面
+with gr.Blocks(title="2048游戏", theme="soft") as demo:
+    gr.Markdown("# 🎮 2048游戏")
+    gr.Markdown("使用方向键或下方的按钮移动方块，相同数字的方块相撞时会合并!")
+    with gr.Row():
+        with gr.Column(scale=2):
+            board_html = gr.HTML(render_board(game.board))
+            with gr.Row(visible=False):
+                status_display = gr.HTML("<b>当前分数:</b> 0<br><b>最大方块:</b> 2")
+        with gr.Column():
+            gr.Markdown("## 手动操作")
+            with gr.Row():
+                gr.Button("上 ↑", elem_id="up-btn").click(
+                    fn=lambda: make_move(0), 
+                    outputs=[board_html, status_display]
+                )
+                gr.Button("左 ←", elem_id="left-btn").click(
+                    fn=lambda: make_move(3), 
+                    outputs=[board_html, status_display]
+                )
+            with gr.Row():
+                gr.Button("下 ↓", elem_id="down-btn").click(
+                    fn=lambda: make_move(2), 
+                    outputs=[board_html, status_display]
+                )
+                gr.Button("右 →", elem_id="right-btn").click(
+                    fn=lambda: make_move(1), 
+                    outputs=[board_html, status_display]
+                )
+            with gr.Row():
+                gr.Button("🔄 重置游戏", elem_id="reset-btn").click(
+                fn=reset_game, 
+                    outputs=[board_html, status_display]
+                )
+            with gr.Row():
+                status_display = gr.HTML("<b>当前分数:</b> 0<br><b>最大方块:</b> 2")
+        with gr.Column():
+            gr.Markdown("## AI操作")
+            gr.Button("🤖 AI移动一步", elem_id="ai-btn").click(
+                fn=ai_move, 
+                outputs=[board_html, status_display]
+            )
+            gr.Markdown("基于DQN神经网络提供支持")
+    
+    # 添加键盘快捷键支持
+    demo.load(
+        fn=None,
+        inputs=None,
+        outputs=None,
+        js="""() => {
+            document.addEventListener('keydown', function(e) {
+                if (e.key === 'ArrowUp') {
+                    document.getElementById('up-btn').click();
+                } else if (e.key === 'ArrowRight') {
+                    document.getElementById('right-btn').click();
+                } else if (e.key === 'ArrowDown') {
+                    document.getElementById('down-btn').click();
+                } else if (e.key === 'ArrowLeft') {
+                    document.getElementById('left-btn').click();
+                } else if (e.key === 'r' || e.key === 'R') {
+                    document.getElementById('reset-btn').click();
+                } else if (e.key === 'a' || e.key === 'A') {
+                    document.getElementById('ai-btn').click();
+                }
+            });
+        }"""
+    )
+    gr.Markdown("### 📚 使用说明")
+    gr.Markdown("1. 使用方向键或下方的按钮移动方块。")
+    gr.Markdown("2. 相同数字的方块相撞时会合并。")
+    gr.Markdown("3. 快捷键说明：上/下/左/右键移动方块，R键重置游戏，A键AI移动一步。")
+    gr.Markdown("4. 点击 '🤖 AI移动一步' 按钮可以使用AI模型进行一步移动。")
+    gr.Markdown("5. 游戏结束后，会显示最终分数和最大方块。")
+
+# 启动界面
+if __name__ == "__main__":
+    demo.launch(share=True)

game2048.py

@@ -0,0 +1,200 @@
+import numpy as np
+import random
+
+class Game2048:
+    def __init__(self, size=4):
+        self.size = size
+        self.reset()
+    
+    def reset(self):
+        """重置游戏状态"""
+        self.board = np.zeros((self.size, self.size), dtype=np.int32)
+        self.score = 0
+        self.add_tile()
+        self.add_tile()
+        self.game_over = False
+        return self.board.copy()
+    
+    def add_tile(self):
+        """在随机空位置添加新方块(90%概率为2,10%概率为4)"""
+        empty_cells = []
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] == 0:
+                    empty_cells.append((i, j))
+        
+        if empty_cells:
+            i, j = random.choice(empty_cells)
+            self.board[i][j] = 2 if random.random() < 0.9 else 4
+    
+    def move(self, direction):
+        """
+        执行移动操作
+        0: 上, 1: 右, 2: 下, 3: 左
+        返回: (新棋盘状态, 游戏是否结束)
+        """
+        moved = False
+        # 根据方向执行移动
+        if direction == 0:  # 上
+            for j in range(self.size):
+                column = self.board[:, j].copy()
+                new_column, moved_col = self.slide(column)
+                if moved_col:
+                    moved = True
+                self.board[:, j] = new_column
+        
+        elif direction == 1:  # 右
+            for i in range(self.size):
+                row = self.board[i, :].copy()[::-1]
+                new_row, moved_row = self.slide(row)
+                if moved_row:
+                    moved = True
+                self.board[i, :] = new_row[::-1]
+        
+        elif direction == 2:  # 下
+            for j in range(self.size):
+                column = self.board[::-1, j].copy()
+                new_column, moved_col = self.slide(column)
+                if moved_col:
+                    moved = True
+                self.board[:, j] = new_column[::-1]
+        
+        elif direction == 3:  # 左
+            for i in range(self.size):
+                row = self.board[i, :].copy()
+                new_row, moved_row = self.slide(row)
+                if moved_row:
+                    moved = True
+                self.board[i, :] = new_row
+        
+        # 如果发生了移动，添加新方块并检查游戏结束
+        if moved:
+            self.add_tile()
+            self.check_game_over()
+        
+        return self.board.copy(), self.game_over
+    
+    def slide(self, line):
+        """处理单行/列的移动和合并逻辑"""
+        non_zero = line[line != 0]
+        new_line = np.zeros_like(line)
+        idx = 0
+        score_inc = 0
+        moved = False
+        
+        # 检查是否移动
+        if not np.array_equal(non_zero, line[:len(non_zero)]):
+            moved = True
+        
+        # 合并相同数字
+        i = 0
+        while i < len(non_zero):
+            if i + 1 < len(non_zero) and non_zero[i] == non_zero[i+1]:
+                new_val = non_zero[i] * 2
+                new_line[idx] = new_val
+                score_inc += new_val
+                i += 2
+                idx += 1
+            else:
+                new_line[idx] = non_zero[i]
+                i += 1
+                idx += 1
+        
+        self.score += score_inc
+        return new_line, moved or (score_inc > 0)
+    
+    def check_game_over(self):
+        """检查游戏是否结束"""
+        # 检查是否还有空格子
+        if np.any(self.board == 0):
+            self.game_over = False
+            return
+        
+        # 检查水平和垂直方向是否有可合并的方块
+        for i in range(self.size):
+            for j in range(self.size - 1):
+                if self.board[i][j] == self.board[i][j+1]:
+                    self.game_over = False
+                    return
+        
+        for j in range(self.size):
+            for i in range(self.size - 1):
+                if self.board[i][j] == self.board[i+1][j]:
+                    self.game_over = False
+                    return
+        
+        self.game_over = True
+    
+    def get_valid_moves(self):
+        """获取当前所有有效移动方向"""
+        valid_moves = []
+        
+        # 检查上移是否有效
+        for j in range(self.size):
+            column = self.board[:, j].copy()
+            new_column, _ = self.slide(column)
+            if not np.array_equal(new_column, self.board[:, j]):
+                valid_moves.append(0)
+                break
+        
+        # 检查右移是否有效
+        for i in range(self.size):
+            row = self.board[i, :].copy()[::-1]
+            new_row, _ = self.slide(row)
+            if not np.array_equal(new_row[::-1], self.board[i, :]):
+                valid_moves.append(1)
+                break
+        
+        # 检查下移是否有效
+        for j in range(self.size):
+            column = self.board[::-1, j].copy()
+            new_column, _ = self.slide(column)
+            if not np.array_equal(new_column[::-1], self.board[:, j]):
+                valid_moves.append(2)
+                break
+        
+        # 检查左移是否有效
+        for i in range(self.size):
+            row = self.board[i, :].copy()
+            new_row, _ = self.slide(row)
+            if not np.array_equal(new_row, self.board[i, :]):
+                valid_moves.append(3)
+                break
+        
+        return valid_moves
+    
+    def get_state(self):
+        """获取当前游戏状态表示（用于AI模型）"""
+        # 创建4个通道的状态表示
+        state = np.zeros((4, self.size, self.size), dtype=np.float32)
+        
+        # 通道0: 当前方块值的对数（归一化）
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] > 0:
+                    state[0, i, j] = np.log2(self.board[i][j]) / 16.0  # 支持到65536 (2^16)
+        
+        # 通道1: 空格子指示器
+        state[1] = (self.board == 0).astype(np.float32)
+        
+        # 通道2: 可合并的邻居指示器
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] > 0:
+                    # 检查右侧
+                    if j < self.size - 1 and self.board[i][j] == self.board[i][j+1]:
+                        state[2, i, j] = 1.0
+                        state[2, i, j+1] = 1.0
+                    # 检查下方
+                    if i < self.size - 1 and self.board[i][j] == self.board[i+1][j]:
+                        state[2, i, j] = 1.0
+                        state[2, i+1, j] = 1.0
+        
+        # 通道3: 最大值位置（归一化）
+        max_value = np.max(self.board)
+        if max_value > 0:
+            max_positions = np.argwhere(self.board == max_value)
+            for pos in max_positions:
+                state[3, pos[0], pos[1]] = 1.0
+        
+        return state

main.py

@@ -0,0 +1,837 @@
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.optim as optim
+import torch.nn.functional as F
+import random
+import os
+from tqdm import tqdm
+import matplotlib.pyplot as plt
+import warnings
+
+if torch.cuda.is_available():
+    device=torch.device("cuda")
+elif torch.xpu.is_available():
+    device=torch.device("xpu")
+else:
+    device=torch.device("cpu")
+print(f"Using device: {device}")
+
+# 2048游戏环境（改进版）
+class Game2048:
+    def __init__(self, size=4):
+        self.size = size
+        self.reset()
+    
+    def reset(self):
+        self.board = np.zeros((self.size, self.size), dtype=np.int32)
+        self.score = 0
+        self.prev_score = 0
+        self.add_tile()
+        self.add_tile()
+        self.game_over = False
+        return self.get_state()
+    
+    def add_tile(self):
+        empty_cells = []
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] == 0:
+                    empty_cells.append((i, j))
+        
+        if empty_cells:
+            i, j = random.choice(empty_cells)
+            self.board[i][j] = 2 if random.random() < 0.9 else 4
+    
+    def move(self, direction):
+        # 0: 上, 1: 右, 2: 下, 3: 左
+        moved = False
+        original_board = self.board.copy()
+        old_score = self.score
+        
+        # 根据方向执行移动
+        if direction == 0:  # 上
+            for j in range(self.size):
+                column = self.board[:, j].copy()
+                new_column, moved_col = self.slide(column)
+                if moved_col:
+                    moved = True
+                self.board[:, j] = new_column
+        
+        elif direction == 1:  # 右
+            for i in range(self.size):
+                row = self.board[i, :].copy()[::-1]
+                new_row, moved_row = self.slide(row)
+                if moved_row:
+                    moved = True
+                self.board[i, :] = new_row[::-1]
+        
+        elif direction == 2:  # 下
+            for j in range(self.size):
+                column = self.board[::-1, j].copy()
+                new_column, moved_col = self.slide(column)
+                if moved_col:
+                    moved = True
+                self.board[:, j] = new_column[::-1]
+        
+        elif direction == 3:  # 左
+            for i in range(self.size):
+                row = self.board[i, :].copy()
+                new_row, moved_row = self.slide(row)
+                if moved_row:
+                    moved = True
+                self.board[i, :] = new_row
+        
+        # 如果发生了移动，添加新方块
+        if moved:
+            self.add_tile()
+            self.check_game_over()
+        
+        reward = self.calculate_reward(old_score, original_board)
+        return self.get_state(), reward, self.game_over
+    
+    def slide(self, line):
+        # 移除零并合并相同数字
+        non_zero = line[line != 0]
+        new_line = np.zeros_like(line)
+        idx = 0
+        score_inc = 0
+        moved = False
+        
+        # 检查是否移动
+        if not np.array_equal(non_zero, line[:len(non_zero)]):
+            moved = True
+        
+        # 合并相同数字
+        i = 0
+        while i < len(non_zero):
+            if i + 1 < len(non_zero) and non_zero[i] == non_zero[i+1]:
+                new_val = non_zero[i] * 2
+                new_line[idx] = new_val
+                score_inc += new_val
+                i += 2
+                idx += 1
+            else:
+                new_line[idx] = non_zero[i]
+                i += 1
+                idx += 1
+        
+        self.score += score_inc
+        return new_line, moved or (score_inc > 0)
+    
+    def calculate_reward(self, old_score, original_board):
+        """改进的奖励函数"""
+        # 1. 基本分数奖励
+        score_reward = (self.score - old_score) * 0.1
+        
+        # 2. 空格子数量变化奖励
+        empty_before = np.count_nonzero(original_board == 0)
+        empty_after = np.count_nonzero(self.board == 0)
+        empty_reward = (empty_after - empty_before) * 0.15
+        
+        # 3. 最大方块奖励
+        max_before = np.max(original_board)
+        max_after = np.max(self.board)
+        max_tile_reward = 0
+        if max_after > max_before:
+            max_tile_reward = np.log2(max_after) * 0.2
+        
+        # 4. 合并奖励（鼓励合并）
+        merge_reward = 0
+        if self.score - old_score > 0:
+            merge_reward = np.log2(self.score - old_score) * 0.1
+        
+        # 5. 单调性惩罚（鼓励有序排列）
+        monotonicity_penalty = self.calculate_monotonicity_penalty() * 0.01
+        
+        # 6. 游戏结束惩罚
+        game_over_penalty = 0
+        if self.game_over:
+            game_over_penalty = -10
+        
+        # 7. 平滑度奖励（鼓励相邻方块值接近）
+        smoothness_reward = self.calculate_smoothness() * 0.01
+        
+        # 总奖励
+        total_reward = (
+            score_reward + 
+            empty_reward + 
+            max_tile_reward + 
+            merge_reward + 
+            smoothness_reward + 
+            monotonicity_penalty + 
+            game_over_penalty
+        )
+        
+        return total_reward
+    
+    def calculate_monotonicity_penalty(self):
+        """计算单调性惩罚（值越低越好）"""
+        penalty = 0
+        for i in range(self.size):
+            for j in range(self.size - 1):
+                if self.board[i][j] > self.board[i][j+1]:
+                    penalty += self.board[i][j] - self.board[i][j+1]
+                else:
+                    penalty += self.board[i][j+1] - self.board[i][j]
+        return penalty
+    
+    def calculate_smoothness(self):
+        """计算平滑度（值越高越好）"""
+        smoothness = 0
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] != 0:
+                    value = np.log2(self.board[i][j])
+                    # 检查右侧邻居
+                    if j < self.size - 1 and self.board[i][j+1] != 0:
+                        neighbor_value = np.log2(self.board[i][j+1])
+                        smoothness -= abs(value - neighbor_value)
+                    # 检查下方邻居
+                    if i < self.size - 1 and self.board[i+1][j] != 0:
+                        neighbor_value = np.log2(self.board[i+1][j])
+                        smoothness -= abs(value - neighbor_value)
+        return smoothness
+    
+    def check_game_over(self):
+        # 检查是否还有空格子
+        if np.any(self.board == 0):
+            self.game_over = False
+            return
+        
+        # 检查水平和垂直方向是否有可合并的方块
+        for i in range(self.size):
+            for j in range(self.size - 1):
+                if self.board[i][j] == self.board[i][j+1]:
+                    self.game_over = False
+                    return
+        
+        for j in range(self.size):
+            for i in range(self.size - 1):
+                if self.board[i][j] == self.board[i+1][j]:
+                    self.game_over = False
+                    return
+        
+        self.game_over = True
+    
+    def get_state(self):
+        """改进的状态表示"""
+        # 创建4个通道的状态表示
+        state = np.zeros((4, self.size, self.size), dtype=np.float32)
+        
+        # 通道0: 当前方块值的对数（归一化）
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] > 0:
+                    state[0, i, j] = np.log2(self.board[i][j]) / 16.0  # 支持到65536 (2^16)
+        
+        # 通道1: 空格子指示器
+        state[1] = (self.board == 0).astype(np.float32)
+        
+        # 通道2: 可合并的邻居指示器
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] > 0:
+                    # 检查右侧
+                    if j < self.size - 1 and self.board[i][j] == self.board[i][j+1]:
+                        state[2, i, j] = 1.0
+                        state[2, i, j+1] = 1.0
+                    # 检查下方
+                    if i < self.size - 1 and self.board[i][j] == self.board[i+1][j]:
+                        state[2, i, j] = 1.0
+                        state[2, i+1, j] = 1.0
+        
+        # 通道3: 最大值位置（归一化）
+        max_value = np.max(self.board)
+        if max_value > 0:
+            max_positions = np.argwhere(self.board == max_value)
+            for pos in max_positions:
+                state[3, pos[0], pos[1]] = 1.0
+        
+        return state
+    
+    def get_valid_moves(self):
+        """更高效的有效移动检测"""
+        valid_moves = []
+        #test_board = np.zeros_like(self.board)
+        
+        # 检查上移是否有效
+        for j in range(self.size):
+            column = self.board[:, j].copy()
+            new_column, _ = self.slide(column)
+            if not np.array_equal(new_column, self.board[:, j]):
+                valid_moves.append(0)
+                break
+        
+        # 检查右移是否有效
+        for i in range(self.size):
+            row = self.board[i, :].copy()[::-1]
+            new_row, _ = self.slide(row)
+            if not np.array_equal(new_row[::-1], self.board[i, :]):
+                valid_moves.append(1)
+                break
+        
+        # 检查下移是否有效
+        for j in range(self.size):
+            column = self.board[::-1, j].copy()
+            new_column, _ = self.slide(column)
+            if not np.array_equal(new_column[::-1], self.board[:, j]):
+                valid_moves.append(2)
+                break
+        
+        # 检查左移是否有效
+        for i in range(self.size):
+            row = self.board[i, :].copy()
+            new_row, _ = self.slide(row)
+            if not np.array_equal(new_row, self.board[i, :]):
+                valid_moves.append(3)
+                break
+        
+        return valid_moves
+
+# 改进的深度Q网络（使用Dueling DQN架构）
+class DQN(nn.Module):
+    def __init__(self, input_channels, output_size):
+        super(DQN, self).__init__()
+        self.input_channels = input_channels
+        
+        # 卷积层
+        self.conv1 = nn.Conv2d(input_channels, 128, kernel_size=3, padding=1)
+        self.conv2 = nn.Conv2d(128, 128, kernel_size=3, padding=1)
+        self.conv3 = nn.Conv2d(128, 128, kernel_size=3, padding=1)
+        
+        # Dueling DQN架构
+        # 价值流
+        self.value_conv = nn.Conv2d(128, 4, kernel_size=1)
+        self.value_fc1 = nn.Linear(4 * 4 * 4, 128)
+        self.value_fc2 = nn.Linear(128, 1)
+        
+        # 优势流
+        self.advantage_conv = nn.Conv2d(128, 16, kernel_size=1)
+        self.advantage_fc1 = nn.Linear(16 * 4 * 4, 128)
+        self.advantage_fc2 = nn.Linear(128, output_size)
+        
+    def forward(self, x):
+        x = F.relu(self.conv1(x))
+        x = F.relu(self.conv2(x))
+        x = F.relu(self.conv3(x))
+        
+        # 价值流
+        value = F.relu(self.value_conv(x))
+        value = value.view(value.size(0), -1)
+        value = F.relu(self.value_fc1(value))
+        value = self.value_fc2(value)
+        
+        # 优势流
+        advantage = F.relu(self.advantage_conv(x))
+        advantage = advantage.view(advantage.size(0), -1)
+        advantage = F.relu(self.advantage_fc1(advantage))
+        advantage = self.advantage_fc2(advantage)
+        
+        # 合并价值流和优势流
+        q_values = value + advantage - advantage.mean(dim=1, keepdim=True)
+        return q_values
+
+# 经验回放缓冲区（带优先级）
+class PrioritizedReplayBuffer:
+    def __init__(self, capacity, alpha=0.6):
+        self.capacity = capacity
+        self.alpha = alpha
+        self.buffer = []
+        self.priorities = np.zeros(capacity)
+        self.pos = 0
+        self.size = 0
+    
+    def push(self, state, action, reward, next_state, done):
+        # 初始优先级设置为最大优先级
+        max_priority = self.priorities.max() if self.buffer else 1.0
+        
+        if len(self.buffer) < self.capacity:
+            self.buffer.append((state, action, reward, next_state, done))
+        else:
+            self.buffer[self.pos] = (state, action, reward, next_state, done)
+        
+        self.priorities[self.pos] = max_priority
+        self.pos = (self.pos + 1) % self.capacity
+        self.size = min(self.size + 1, self.capacity)
+    
+    def sample(self, batch_size, beta=0.4):
+        if self.size == 0:
+            return None, None, None
+        
+        priorities = self.priorities[:self.size]
+        probs = priorities ** self.alpha
+        probs /= probs.sum()
+        
+        indices = np.random.choice(self.size, batch_size, p=probs)
+        samples = [self.buffer[idx] for idx in indices]
+        
+        # 计算重要性采样权重
+        weights = (self.size * probs[indices]) ** (-beta)
+        weights /= weights.max()
+        weights = np.array(weights, dtype=np.float32)
+        
+        states, actions, rewards, next_states, dones = zip(*samples)
+        return (
+            torch.tensor(np.array(states)), 
+            torch.tensor(actions, dtype=torch.long), 
+            torch.tensor(rewards, dtype=torch.float),
+            torch.tensor(np.array(next_states)),
+            torch.tensor(dones, dtype=torch.float),
+            indices,
+            torch.tensor(weights)
+        )
+    
+    def update_priorities(self, indices, priorities):
+        # 确保 priorities 是一个数组
+        if isinstance(priorities, np.ndarray) and priorities.ndim == 1:
+            for idx, priority in zip(indices, priorities):
+                self.priorities[idx] = priority
+        else:
+            # 处理标量情况（虽然不应该发生）
+            if not isinstance(priorities, (list, np.ndarray)):
+                priorities = [priorities] * len(indices)
+            for idx, priority in zip(indices, priorities):
+                self.priorities[idx] = priority
+    
+    def __len__(self):
+        return self.size
+
+# 改进的DQN智能体
+class DQNAgent:
+    def __init__(self, input_channels, action_size, lr=3e-4, gamma=0.99, 
+                 epsilon_start=1.0, epsilon_end=0.01, epsilon_decay=0.999, 
+                 target_update_freq=1000, batch_size=128):
+        self.input_channels = input_channels
+        self.action_size = action_size
+        self.gamma = gamma
+        self.epsilon = epsilon_start
+        self.epsilon_end = epsilon_end
+        self.epsilon_decay = epsilon_decay
+        self.batch_size = batch_size
+        self.target_update_freq = target_update_freq
+        
+        # 主网络和目标网络
+        self.policy_net = DQN(input_channels, action_size).to(device)
+        self.target_net = DQN(input_channels, action_size).to(device)
+        self.target_net.load_state_dict(self.policy_net.state_dict())
+        self.target_net.eval()
+        
+        self.optimizer = optim.Adam(self.policy_net.parameters(), lr=lr, weight_decay=1e-5)
+        self.memory = PrioritizedReplayBuffer(50000)
+        self.steps_done = 0
+        self.loss_fn = nn.SmoothL1Loss(reduction='none')
+    
+    def select_action(self, state, valid_moves):
+        self.steps_done += 1
+        self.epsilon = max(self.epsilon_end, self.epsilon * self.epsilon_decay)
+        
+        if random.random() < self.epsilon:
+            # 随机选择有效动作
+            return random.choice(valid_moves)
+        else:
+            # 使用策略网络选择动作
+            with torch.no_grad():
+                state_tensor = torch.tensor(state, dtype=torch.float).unsqueeze(0).to(device)
+                q_values = self.policy_net(state_tensor).cpu().numpy().flatten()
+                
+                # 只考虑有效动作
+                valid_q_values = np.full(self.action_size, -np.inf)
+                for move in valid_moves:
+                    valid_q_values[move] = q_values[move]
+                
+                return np.argmax(valid_q_values)
+    
+    def optimize_model(self, beta=0.4):
+        if len(self.memory) < self.batch_size:
+            return 0
+        
+        # 从回放缓冲区采样
+        sample = self.memory.sample(self.batch_size, beta)
+        if sample is None:
+            return 0
+            
+        states, actions, rewards, next_states, dones, indices, weights = sample
+        
+        states = states.to(device)
+        actions = actions.to(device)
+        rewards = rewards.to(device)
+        next_states = next_states.to(device)
+        dones = dones.to(device)
+        weights = weights.to(device)
+        
+        # 计算当前Q值
+        current_q = self.policy_net(states).gather(1, actions.unsqueeze(1)).squeeze()
+        
+        # 计算目标Q值（Double DQN）
+        with torch.no_grad():
+            next_actions = self.policy_net(next_states).max(1)[1]
+            next_q = self.target_net(next_states).gather(1, next_actions.unsqueeze(1)).squeeze()
+            target_q = rewards + (1 - dones) * self.gamma * next_q
+        
+        # 计算损失
+        losses = self.loss_fn(current_q, target_q)
+        loss = (losses * weights).mean()
+        
+        # 更新优先级（使用每个样本的损失绝对值）
+        with torch.no_grad():
+            priorities = losses.abs().cpu().numpy() + 1e-5
+            self.memory.update_priorities(indices, priorities)
+        
+        # 优化模型
+        self.optimizer.zero_grad()
+        loss.backward()
+        
+        # 梯度裁剪
+        torch.nn.utils.clip_grad_norm_(self.policy_net.parameters(), 10)
+        
+        self.optimizer.step()
+        
+        return loss.item()
+    
+    def update_target_network(self):
+        self.target_net.load_state_dict(self.policy_net.state_dict())
+    
+    def save_model(self, path):
+        torch.save({
+            'policy_net_state_dict': self.policy_net.state_dict(),
+            'target_net_state_dict': self.target_net.state_dict(),
+            'optimizer_state_dict': self.optimizer.state_dict(),
+            'epsilon': self.epsilon,
+            'steps_done': self.steps_done
+        }, path)
+    
+    def load_model(self, path):
+        if not os.path.exists(path):
+            print(f"Model file not found: {path}")
+            return
+            
+        try:
+            # 尝试使用 weights_only=False 加载模型
+            checkpoint = torch.load(path, map_location=device, weights_only=False)
+            self.policy_net.load_state_dict(checkpoint['policy_net_state_dict'])
+            self.target_net.load_state_dict(checkpoint['target_net_state_dict'])
+            self.optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
+            self.epsilon = checkpoint['epsilon']
+            self.steps_done = checkpoint['steps_done']
+            self.policy_net.eval()
+            self.target_net.eval()
+            print(f"Model loaded successfully from {path}")
+        except Exception as e:
+            print(f"Error loading model: {e}")
+            # 尝试使用旧版加载方式作为备选
+            try:
+                warnings.warn("Trying legacy load method without weights_only")
+                checkpoint = torch.load(path, map_location=device)
+                self.policy_net.load_state_dict(checkpoint['policy_net_state_dict'])
+                self.target_net.load_state_dict(checkpoint['target_net_state_dict'])
+                self.optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
+                self.epsilon = checkpoint['epsilon']
+                self.steps_done = checkpoint['steps_done']
+                self.policy_net.eval()
+                self.target_net.eval()
+                print(f"Model loaded successfully using legacy method")
+            except Exception as e2:
+                print(f"Failed to load model: {e2}")
+# 训练函数（带进度记录）
+def train_agent(agent, env, episodes=5000, save_path='models/dqn_2048.pth', 
+                checkpoint_path='models/checkpoint.pth', resume=False, start_episode=0):
+    # 创建保存模型的目录
+    os.makedirs(os.path.dirname(save_path), exist_ok=True)
+    
+    # 记录训练指标
+    scores = []
+    max_tiles = []
+    avg_scores = []
+    losses = []
+    best_score = 0
+    best_max_tile = 0
+    
+    # 如果续训，加载训练状态
+    if resume and os.path.exists(checkpoint_path):
+        try:
+            # 使用 weights_only=False 加载检查点
+            checkpoint = torch.load(checkpoint_path, map_location=device, weights_only=False)
+            scores = checkpoint['scores']
+            max_tiles = checkpoint['max_tiles']
+            avg_scores = checkpoint['avg_scores']
+            losses = checkpoint['losses']
+            best_score = checkpoint.get('best_score', 0)
+            best_max_tile = checkpoint.get('best_max_tile', 0)
+            print(f"Resuming training from episode {start_episode}...")
+        except Exception as e:
+            print(f"Error loading checkpoint: {e}")
+            print("Starting training from scratch...")
+            resume = False
+    
+    if not resume:
+        start_episode = 0
+    
+    # 使用tqdm显示进度条
+    progress_bar = tqdm(range(start_episode, episodes), desc="Training")
+    
+    for episode in progress_bar:
+        state = env.reset()
+        total_reward = 0
+        done = False
+        steps = 0
+        episode_loss = 0
+        loss_count = 0
+        
+        while not done:
+            valid_moves = env.get_valid_moves()
+            if not valid_moves:
+                done = True
+                continue
+                
+            action = agent.select_action(state, valid_moves)
+            next_state, reward, done = env.move(action)
+            total_reward += reward
+            
+            agent.memory.push(state, action, reward, next_state, done)
+            state = next_state
+            
+            # 优化模型
+            loss = agent.optimize_model(beta=min(1.0, episode / 1000))
+            if loss > 0:
+                episode_loss += loss
+                loss_count += 1
+            
+            # 定期更新目标网络
+            if agent.steps_done % agent.target_update_freq == 0:
+                agent.update_target_network()
+            
+            steps += 1
+        
+        # 记录分数和最大方块
+        score = env.score
+        max_tile = np.max(env.board)
+        scores.append(score)
+        max_tiles.append(max_tile)
+        
+        # 计算平均损失
+        avg_loss = episode_loss / loss_count if loss_count > 0 else 0
+        losses.append(avg_loss)
+        
+        # 更新最佳记录
+        if score > best_score:
+            best_score = score
+            agent.save_model(save_path.replace('.pth', '_best_score.pth'))
+        if max_tile > best_max_tile:
+            best_max_tile = max_tile
+            agent.save_model(save_path.replace('.pth', '_best_tile.pth'))
+        
+        # 计算最近100轮平均分数
+        recent_scores = scores[-100:] if len(scores) >= 100 else scores
+        avg_score = np.mean(recent_scores)
+        avg_scores.append(avg_score)
+        
+        # 更新进度条描述
+        progress_bar.set_description(
+            f"Ep {episode+1}/{episodes} | "
+            f"Score: {score} (Avg: {avg_score:.1f}) | "
+            f"Max Tile: {max_tile} | "
+            f"Loss: {avg_loss:.4f} | "
+            f"Epsilon: {agent.epsilon:.4f}"
+        )
+        
+        # 定期保存模型和训练状态
+        if (episode + 1) % 100 == 0:
+            agent.save_model(save_path)
+            
+            # 保存训练状态
+            checkpoint = {
+                'scores': scores,
+                'max_tiles': max_tiles,
+                'avg_scores': avg_scores,
+                'losses': losses,
+                'best_score': best_score,
+                'best_max_tile': best_max_tile,
+                'episode': episode + 1,
+                'steps_done': agent.steps_done,
+                'epsilon': agent.epsilon
+            }
+            try:
+                torch.save(checkpoint, checkpoint_path)
+            except Exception as e:
+                print(f"Error saving checkpoint: {e}")
+            
+            # 绘制训练曲线
+            if episode > 100:  # 确保有足够的数据
+                plt.figure(figsize=(12, 8))
+                
+                # 分数曲线
+                plt.subplot(2, 2, 1)
+                plt.plot(scores, label='Score')
+                plt.plot(avg_scores, label='Avg Score (100 eps)')
+                plt.xlabel('Episode')
+                plt.ylabel('Score')
+                plt.title('Training Scores')
+                plt.legend()
+                
+                # 最大方块曲线
+                plt.subplot(2, 2, 2)
+                plt.plot(max_tiles, 'g-')
+                plt.xlabel('Episode')
+                plt.ylabel('Max Tile')
+                plt.title('Max Tile Achieved')
+                
+                # 损失曲线
+                plt.subplot(2, 2, 3)
+                plt.plot(losses, 'r-')
+                plt.xlabel('Episode')
+                plt.ylabel('Loss')
+                plt.title('Training Loss')
+                
+                # 分数分布直方图
+                plt.subplot(2, 2, 4)
+                plt.hist(scores, bins=20, alpha=0.7)
+                plt.xlabel('Score')
+                plt.ylabel('Frequency')
+                plt.title('Score Distribution')
+                
+                plt.tight_layout()
+                plt.savefig('training_progress.png')
+                plt.close()
+    
+    # 保存最终模型
+    agent.save_model(save_path)
+    
+    return scores, max_tiles, losses
+# 推理函数（带可视化）
+def play_with_model(agent, env, episodes=3):
+    agent.epsilon = 0.001  # 设置很小的epsilon值进行推理
+    
+    for episode in range(episodes):
+        state = env.reset()
+        done = False
+        steps = 0
+        
+        print(f"\nEpisode {episode+1}")
+        print("Initial Board:")
+        print(env.board)
+        
+        while not done:
+            valid_moves = env.get_valid_moves()
+            if not valid_moves:
+                done = True
+                print("No valid moves left!")
+                continue
+                
+            # 选择动作
+            with torch.no_grad():
+                state_tensor = torch.tensor(state, dtype=torch.float).unsqueeze(0).to(device)
+                q_values = agent.policy_net(state_tensor).cpu().numpy().flatten()
+                
+                # 只考虑有效动作
+                valid_q_values = np.full(env.size, -np.inf)
+                for move in valid_moves:
+                    valid_q_values[move] = q_values[move]
+                
+                action = np.argmax(valid_q_values)
+            
+            # 执行动作
+            next_state, reward, done = env.move(action)
+            state = next_state
+            steps += 1
+            
+            # 渲染游戏
+            print(f"\nStep {steps}: Action {['Up', 'Right', 'Down', 'Left'][action]}")
+            print(env.board)
+            print(f"Score: {env.score}, Max Tile: {np.max(env.board)}")
+            #同时将结果保存至result.txt文件中
+            with open("result.txt", "a") as f:
+                f.write(f"Episode {episode+1}, Step {steps}, Action {['Up', 'Right', 'Down', 'Left'][action]}, Score: {env.score}, Max Tile: {np.max(env.board)}\n{env.board}\n")
+            f.close()
+
+        
+        print(f"\nGame Over! Final Score: {env.score}, Max Tile: {np.max(env.board)}")
+
+# 主程序
+if __name__ == "__main__":
+    args = {"train":0, "resume":0, "play":1, "episodes":50000}
+    env = Game2048(size=4)
+    input_channels = 4  # 状态表示的通道数
+    action_size = 4  # 上、右、下、左
+    
+    agent = DQNAgent(
+        input_channels, 
+        action_size,
+        lr=1e-4,
+        epsilon_decay=0.999,  # 更慢的衰减
+        target_update_freq=1000,
+        batch_size=256
+    )
+    
+    # 训练模型
+    if args.get('train') or args.get('resume'):
+        print("Starting training...")
+        
+        # 如果续训，加载检查点
+        start_episode = 0
+        checkpoint_path = 'models/checkpoint.pth'
+        if args.get('resume') and os.path.exists(checkpoint_path):
+            try:
+                # 使用 weights_only=False 加载检查点
+                checkpoint = torch.load(checkpoint_path, map_location=device, weights_only=False)
+                start_episode = checkpoint.get('episode', 0)
+                agent.steps_done = checkpoint.get('steps_done', 0)
+                agent.epsilon = checkpoint.get('epsilon', agent.epsilon)
+            except Exception as e:
+                print(f"Error loading checkpoint: {e}")
+                print("Starting training from scratch...")
+                start_episode = 0
+        
+        scores, max_tiles, losses = train_agent(
+            agent, 
+            env, 
+            episodes=args.get('episodes'),
+            save_path='models/dqn_2048.pth',
+            checkpoint_path=checkpoint_path,
+            resume=args.get('resume'),
+            start_episode=start_episode
+        )
+        print("Training completed!")
+        
+        # 绘制最终训练结果
+        plt.figure(figsize=(15, 10))
+        
+        plt.subplot(3, 1, 1)
+        plt.plot(scores)
+        plt.title('Scores per Episode')
+        plt.xlabel('Episode')
+        plt.ylabel('Score')
+        
+        plt.subplot(3, 1, 2)
+        plt.plot(max_tiles)
+        plt.title('Max Tile per Episode')
+        plt.xlabel('Episode')
+        plt.ylabel('Max Tile')
+        
+        plt.subplot(3, 1, 3)
+        plt.plot(losses)
+        plt.title('Training Loss per Episode')
+        plt.xlabel('Episode')
+        plt.ylabel('Loss')
+        
+        plt.tight_layout()
+        plt.savefig('final_training_results.png')
+        plt.close()
+    
+    # 加载模型并推理
+    if args.get('play'):
+        model_path = 'models/dqn_2048_best_tile.pth'
+        if not os.path.exists(model_path):
+            model_path = 'models/dqn_2048.pth'
+        
+        if os.path.exists(model_path):
+            agent.load_model(model_path)
+            print("Playing with trained model...")
+            if not os.path.exists("result.txt"):
+                play_with_model(agent, env, episodes=1)
+            else:
+                os.remove("result.txt")            #删除之前记录
+                play_with_model(agent, env, episodes=1)
+
+        else:
+            print("No trained model found. Please train the model first.")