Upload 13 files

PLAN.md

@@ -0,0 +1,230 @@
+# 项目计划：基于Transformer的2048游戏AI
+
+## 技术方案概述
+
+### 硬件约束
+- AMD Ryzen 5 PRO 4650U (6核12线程) CPU
+- 8GB RAM
+- 无NVIDIA GPU，纯CPU训练
+- 需要小型高效的模型架构
+
+### 整体架构
+
+```
+┌─────────────────────────────────────────────────────────┐
+│                    GUI主窗口 (PyQt5)                      │
+├─────────────────────────────────────────────────────────┤
+│  ┌──────────────┐  ┌──────────────────────────────────┐ │
+│  │  2048游戏面板 │  │         训练状态面板              │ │
+│  │   (4x4网格)   │  │  - 当前分数/局面分数              │ │
+│  │              │  │  - 训练速度 (games/sec)          │ │
+│  │              │  │  - 累积分数变化曲线               │ │
+│  │              │  │  - 局面分数变化曲线               │ │
+│  └──────────────┘  └──────────────────────────────────┘ │
+├─────────────────────────────────────────────────────────┤
+│  控制面板: [训练模式] [演示模式] [开始/停止] [AI托管]     │
+└─────────────────────────────────────────────────────────┘
+```
+
+## 模块设计
+
+### 1. 游戏核心模块 (`game.py`)
+
+```python
+class Game2048:
+    """2048游戏核心逻辑"""
+    
+    def __init__(self):
+        self.board: np.ndarray  # 4x4棋盘
+        self.accumulated_score: int  # 累积分数
+        self.situational_score: float  # 局面分数
+        
+    def reset(self) -> None: ...
+    def move(self, direction: int) -> tuple[bool, bool]: ...  # (moved, game_over)
+    def get_state(self) -> np.ndarray: ...  # 返回当前局面
+    def calculate_situational_score(self) -> float: ...
+```
+
+**局面分数计算公式：**
+```
+situation_score = (
+    empty_cells * 10 +                           # 空格越多越好
+    max_consecutive_adjacent * 15 +              # 连续相邻数字越多越好
+    log2(max_tile) * 5 -                         # 最高数字的对数
+    monotonicity_penalty                         # 单调性惩罚（避免混乱）
+)
+```
+
+### 2. Transformer模型 (`model.py`)
+
+考虑到CPU训练的限制，采用小型Transformer：
+
+```python
+class Game2048Transformer(nn.Module):
+    """小型Transformer用于2048决策"""
+    
+    def __init__(self):
+        # 输入: 4x4棋盘 + 2个分数特征
+        # 将棋盘展平为16个token，每个token代表一个格子的状态
+        
+        self.embedding = nn.Embedding(16, 64)  # 0-15 表示 log2(value)，16表示空
+        self.score_embedding = nn.Linear(2, 64)  # 两种分数的embedding
+        
+        encoder_layer = nn.TransformerEncoderLayer(
+            d_model=64,
+            nhead=4,
+            dim_feedforward=128,
+            dropout=0.1,
+            batch_first=True
+        )
+        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=2)
+        
+        self.policy_head = nn.Linear(64, 4)   # 输出4个动作的概率
+        self.value_head = nn.Linear(64, 1)    # 输出状态价值
+```
+
+**模型大小估算：**
+- Embedding: 17 * 64 = 1,088 参数
+- Transformer (2层): ~50,000 参数
+- 输出头: ~300 参数
+- **总计: ~52,000 参数** - 非常小，适合CPU训练
+
+### 3. 训练模块 (`trainer.py`)
+
+采用 **Actor-Critic + PPO** 策略：
+
+```python
+class PPOTrainer:
+    """PPO训练器"""
+    
+    def __init__(self, model, lr=1e-4):
+        self.model = model
+        self.optimizer = torch.optim.Adam(model.parameters(), lr=lr)
+        
+    def compute_advantage(self, rewards, values, dones):
+        # 计算GAE (Generalized Advantage Estimation)
+        ...
+        
+    def update(self, trajectories):
+        # PPO更新逻辑
+        ...
+```
+
+**奖励设计：**
+```python
+reward = (
+    accumulated_score_delta * 0.3 +     # 累积分数增量（权重低）
+    situational_score * 0.7 +           # 局面分数（权重高）
+    game_over_penalty * (-100)          # 游戏结束惩罚
+)
+```
+
+### 4. 多进程训练 (`parallel.py`)
+
+利用6核CPU，同时运行多个游戏实例：
+
+```python
+class ParallelGameEnv:
+    """并行游戏环境"""
+    
+    def __init__(self, num_envs=4):
+        self.num_envs = num_envs
+        self.envs = [Game2048() for _ in range(num_envs)]
+        
+    def step(self, actions: list[int]) -> list[Transition]:
+        # 并行执行动作，返回状态转移
+        ...
+```
+
+### 5. GUI模块 (`gui.py`)
+
+使用 PyQt5 构建界面：
+
+```python
+class MainWindow(QMainWindow):
+    """主窗口"""
+    
+    def __init__(self):
+        self.game_widget = GameBoardWidget()
+        self.stats_widget = StatsWidget()
+        self.control_widget = ControlWidget()
+        
+        # 训练线程
+        self.training_thread = TrainingThread()
+        
+    def switch_mode(self, mode: str): ...
+    def update_display(self): ...
+```
+
+## 文件结构
+
+```
+game2048/
+├── TASK.md          # 任务描述
+├── PLAN.md          # 本文件
+├── main.py          # 入口文件
+├── game.py          # 游戏核心逻辑
+├── model.py         # Transformer模型定义
+├── trainer.py       # PPO训练器
+├── parallel.py      # 多进程训练
+├── gui.py           # GUI界面
+├── utils.py         # 工具函数
+├── requirements.txt # 依赖
+└── checkpoints/     # 模型保存目录
+```
+
+## 实现步骤
+
+### 阶段1: 核心游戏逻辑
+1. 实现 `game.py` - 2048游戏规则
+2. 实现局面分数计算
+3. 编写游戏逻辑单元测试
+
+### 阶段2: 模型与训练
+4. 实现 `model.py` - Transformer模型
+5. 实现 `trainer.py` - PPO训练器
+6. 实现 `parallel.py` - 多进程环境
+7. 验证训练流程可以运行
+
+### 阶段3: GUI界面
+8. 实现 `gui.py` - 主窗口和游戏面板
+9. 实现训练状态可视化（分数曲线）
+10. 实现模式切换（训练/演示）
+
+### 阶段4: 整合与优化
+11. 整合所有模块
+12. 性能优化
+13. 模型保存/加载功能
+
+## 依赖
+
+```
+torch>=2.0.0
+numpy>=1.24.0
+PyQt5>=5.15.0
+matplotlib>=3.7.0
+```
+
+## 训练策略细节
+
+### 状态表示
+- 棋盘状态：将每个格子的值转换为 log2(value)，空格为0
+- 分数归一化：累积分数和局面分数归一化到 [0, 1]
+
+### 动作空间
+- 0: 上
+- 1: 下  
+- 2: 左
+- 3: 右
+
+### 训练超参数
+- Learning rate: 1e-4
+- Batch size: 64
+- PPO clip ratio: 0.2
+- GAE lambda: 0.95
+- Discount factor (gamma): 0.99
+- 并行环境数: 4 (根据CPU核心数调整)
+
+### 停止条件
+- 连续100局游戏平均分数无提升
+- 用户手动停止

README.md

@@ -0,0 +1,318 @@
+# 2048 AI Trainer
+
+基于 Transformer 的 2048 游戏人工智能训练器，使用 PPO（Proximal Policy Optimization）强化学习算法，让 AI 学会玩 2048 游戏。
+
+## 项目简介
+
+本项目实现了一个完整的 2048 游戏 AI 训练系统，包括：
+
+- **游戏引擎**: 完整的 2048 游戏逻辑实现
+- **深度学习模型**: 基于 Transformer 架构的策略网络
+- **强化学习训练**: PPO 算法实现
+- **可视化界面**: PyQt5 图形界面，支持训练监控和演示
+- **命令行工具**: 支持无 GUI 的训练和演示模式
+
+## 功能特点
+
+### 1. Transformer 模型
+
+采用小型 Transformer 架构，专为 CPU 训练优化：
+
+- **参数量**: 约 77,000 个参数（~300KB）
+- **输入处理**: 
+  - 棋盘状态编码为 16 个 token（每个格子对应一个 token）
+  - 分数特征（累积分数、局面分数）作为额外输入
+  - 位置编码：行/列位置嵌入
+- **网络结构**:
+  - 2 层 Transformer Encoder
+  - 4 个注意力头
+  - 隐藏维度 64
+  - 前馈网络维度 128
+- **输出**:
+  - 策略头：4 个动作（上/下/左/右）的概率分布
+  - 价值头：当前状态的价值评估
+
+### 2. 双评分机制
+
+#### 累积分数（Accumulated Score）
+传统 2048 计分方式，每次合成砖块获得合成后砖块的数值作为分数。
+
+#### 局面分数（Situational Score）
+综合评估当前局面的质量，鼓励 AI 保持良好局面：
+
+```
+局面分数 = 空格数 × 10 + 最大连续相邻数 × 15 + log₂(最大砖块) × 5 + 单调性奖励
+```
+
+- **空格数**: 空格越多，操作空间越大
+- **连续相邻数**: 如 512-1024-2048 连续排列，便于后续合并
+- **单调性**: 鼓励数字按方向有序排列
+
+### 3. PPO 训练算法
+
+使用 Proximal Policy Optimization 算法进行训练：
+
+- **优势估计**: GAE（Generalized Advantage Estimation）
+- **策略裁剪**: 防止策略更新过大
+- **价值函数**: 辅助训练，提供状态价值估计
+- **熵正则化**: 鼓励探索
+
+### 4. GUI 界面
+
+基于 PyQt5 的图形界面：
+
+- **训练模式**: 
+  - 设置训练局数
+  - 实时显示训练进度
+  - 分数曲线可视化
+  - 训练完成后自动保存模型
+  
+- **演示模式**:
+  - 键盘手动操作
+  - AI 托管模式
+  - 单步执行
+  - 自动连续执行
+  - 实时局面分数曲线
+
+## 安装
+
+### 环境要求
+
+- Python 3.8+
+- Windows / Linux / macOS
+
+### 安装依赖
+
+```bash
+cd game2048
+pip install -r requirements.txt
+```
+
+### 依赖列表
+
+```
+torch>=2.0.0      # 深度学习框架
+numpy<2           # 数值计算
+PyQt5>=5.15.0     # GUI 框架
+matplotlib>=3.7.0 # 绘图库
+```
+
+## 使用方法
+
+### 1. GUI 模式
+
+```bash
+python main.py
+```
+
+启动图形界面后：
+
+**训练模式**:
+1. 选择 "Training Mode"
+2. 设置训练局数（默认 500）
+3. 点击 "Start Training" 开始训练
+4. 训练完成后自动保存到 `checkpoints/model.pt`
+5. 可随时点击 "Stop Training" 停止
+
+**演示模式**:
+1. 选择 "Demo Mode"
+2. 点击 "Load Model" 加载已训练模型
+3. 使用方式：
+   - 键盘方向键：手动操作
+   - "AI Mode"：切换 AI 托管
+   - "Step"：AI 单步执行
+   - "Auto"：AI 自动连续执行
+   - "Reset"：重新开始游戏
+
+### 2. 命令行训练
+
+```bash
+# 训练 1000 局
+python main.py --train --games 1000
+
+# 使用 4 个并行环境
+python main.py --train --games 1000 --envs 4
+
+# 设置随机种子
+python main.py --train --games 1000 --seed 42
+```
+
+### 3. 演示模式
+
+```bash
+# 加载模型并演示 5 局
+python main.py --demo --model checkpoints/model.pt --games 5
+
+# 不加载模型（随机权重）
+python main.py --demo --games 3
+```
+
+### 4. 简单训练脚本
+
+```bash
+python train_simple.py
+```
+
+修改脚本末尾可调整训练参数：
+
+```python
+train_simple(num_games=500, save_path="checkpoints/model.pt")
+```
+
+## 项目结构
+
+```
+game2048/
+├── TASK.md              # 任务需求文档
+├── PLAN.md              # 项目计划文档
+├── README.md            # 本文件
+├── main.py              # 程序入口
+├── game.py              # 2048 游戏核心逻辑
+│   ├── Game2048         # 游戏类
+│   ├── move()           # 移动操作
+│   ├── get_state()      # 获取状态
+│   └── calculate_situational_score()  # 计算局面分数
+│
+├── model.py             # Transformer 模型
+│   ├── Game2048Transformer  # Transformer 模型
+│   ├── Game2048CNN          # CNN 备选模型
+│   └── get_action()         # 动作选择
+│
+├── trainer.py           # PPO 训练器
+│   ├── PPOTrainer       # PPO 训练类
+│   ├── RolloutBuffer    # 经验缓冲区
+│   ├── Transition       # 状态转移数据结构
+│   └── TrainingStats    # 训练统计
+│
+├── parallel.py          # 并行训练环境
+│   ├── ParallelGameEnv  # 并行游戏环境
+│   ├── TrainingWorker   # 训练工作器
+│   └── TrainingLoop     # 训练循环
+│
+├── gui.py               # GUI 界面
+│   ├── MainWindow       # 主窗口
+│   ├── GameBoardWidget  # 游戏面板
+│   ├── ScoreWidget      # 分数显示
+│   ├── PlotCanvas       # 曲线绑图
+│   └── SimpleTrainingThread  # 训练线程
+│
+├── train_simple.py      # 简化训练脚本
+├── utils.py             # 工具函数
+├── requirements.txt     # 依赖列表
+└── checkpoints/         # 模型保存目录
+    └── model.pt         # 训练好的模型
+```
+
+## 模型架构详解
+
+### 输入表示
+
+```python
+# 棋盘状态 (4, 4)
+# 每个格子值转换为 log₂(value)，空格为 0
+state = [[0, 1, 2, 0],   # 对应 [空, 2, 4, 空]
+         [1, 2, 3, 1],   # 对应 [2, 4, 8, 2]
+         ...]
+
+# 分数特征 (2,)
+# [归一化累积分数, 归一化局面分数]
+scores = [0.05, 0.85]
+```
+
+### 网络结构
+
+```
+Input: (batch, 4, 4) board + (batch, 2) scores
+    ↓
+Position Embedding: (batch, 16, 64)
+    + Spatial Embedding: (batch, 16, 64)
+    + Score Embedding: (batch, 1, 64)
+    ↓
+Transformer Encoder (2 layers)
+    - Multi-Head Attention (4 heads)
+    - Feed-Forward Network (dim=128)
+    ↓
+Global Mean Pooling: (batch, 64)
+    ↓
+    ├── Policy Head → (batch, 4)  # 动作概率
+    └── Value Head  → (batch, 1)  # 状态价值
+```
+
+## 训练策略
+
+### 奖励设计
+
+```python
+reward = 局面分数变化 × 0.7 + 累积分数增量 × 0.003
+
+# 游戏结束惩罚
+if game_over:
+    reward -= 10.0
+```
+
+### 超参数
+
+| 参数 | 值 |
+|------|-----|
+| Learning Rate | 3e-4 |
+| Batch Size | 64 |
+| PPO Clip Ratio | 0.2 |
+| GAE Lambda | 0.95 |
+| Discount Factor (γ) | 0.99 |
+| Entropy Coefficient | 0.01 |
+
+## 训练结果
+
+### 500 局训练后
+
+| 指标 | 数值 |
+|------|------|
+| 平均分数 | ~2500 |
+| 最高分数 | 6812 |
+| 最大砖块 | 512 |
+| 训练时间 | ~9 分钟 |
+
+### 分数分布
+
+```
+随机权重: 平均 ~800, 最高 ~2000
+训练 500 局: 平均 ~2500, 最高 ~6800
+```
+
+## 开发说明
+
+### 添加新功能
+
+1. **修改局面评分**: 编辑 `game.py` 中的 `calculate_situational_score()`
+2. **调整模型**: 修改 `model.py` 中的网络结构
+3. **优化训练**: 调整 `trainer.py` 中的超参数
+
+### 调试模式
+
+```python
+# 在 game.py 中测试游戏逻辑
+if __name__ == "__main__":
+    game = Game2048()
+    print(game)
+    game.move(Game2048.LEFT)
+    print(game)
+```
+
+## 已知问题
+
+- Windows 下 PyTorch 可能需要特定版本以避免 DLL 加载问题
+- NumPy 2.x 与 PyTorch 存在兼容性问题，建议使用 NumPy < 2
+
+## 参考资料
+
+- [PPO 论文](https://arxiv.org/abs/1707.06347)
+- [Transformer 论文](https://arxiv.org/abs/1706.03762)
+- [2048 游戏](https://play2048.co/)
+
+## 许可证
+
+MIT License
+
+---
+
+*本项目由 GLM-5 开发实现*

TASK.md

@@ -0,0 +1,24 @@
+# 需求
+
+用Python制作一个基于Transformer的2048游戏AI
+
+可以创建一个简单的窗口让我看到实时的训练状态，包括训练时AI的当前分数、训练速度、历史已经结束的游戏的*累积分数*变化、当前游戏的*局面分数*变化，窗口可切换演示模式和训练模式
+
+2048的一部分规定补充：
+
+- 开局会在随机位置生成一个2，后续在空格生成的砖块在随机位置且值为2或4
+- 同时存在两个分数机制：累积分数和局面分数
+- 累积分数机制：每合成一个增加合成后的砖块的数值
+- 局面分数机制：根据*空格数量*、*相邻连续数字数量（相邻砖块拥有相邻数字的数量，取最大值，不计算重复，如512 1024 2048为3，256 512 512 2为2）*、*最高数字的对数*设计以恶合理的函数，局面分数越高越好
+
+## 第一步：训练模型
+
+为模型提供当前的局面状态和两种分数（累积分数的权重低于其它，目的是局面分数保持更高），输出只有上下左右
+
+游戏结束后再开始新的游戏继续训练，直到训练没有长进或我手动点击停止
+
+性能允许的话可以多开2048游戏进行训练
+
+## 第二步：演示模式
+
+在窗口中可以手动操作或AI托管，AI托管可以单步执行也可以持续下一步，可以看到两种分数及局面分数变化

checkpoints/model_1500.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:11ff8939d4676c1ad76ca9c5b0ec2616bfc48ed10788c36ff2b0e50cd7fbcb35
+size 415458

game.py

@@ -0,0 +1,388 @@
+"""
+2048游戏核心逻辑
+"""
+import numpy as np
+from typing import Tuple, Optional
+import random
+
+
+class Game2048:
+    """2048游戏核心类"""
+    
+    # 动作定义
+    UP = 0
+    DOWN = 1
+    LEFT = 2
+    RIGHT = 3
+    
+    def __init__(self):
+        self.board: np.ndarray = np.zeros((4, 4), dtype=np.int64)
+        self.accumulated_score: int = 0
+        self.situational_score: float = 0.0
+        self.game_over: bool = False
+        self.moves_count: int = 0
+        self.reset()
+    
+    def reset(self) -> np.ndarray:
+        """重置游戏，返回初始状态"""
+        self.board = np.zeros((4, 4), dtype=np.int64)
+        self.accumulated_score = 0
+        self.situational_score = 0.0
+        self.game_over = False
+        self.moves_count = 0
+        
+        # 开局生成一个2
+        self._spawn_tile(value=2)
+        self._update_situational_score()
+        return self.get_state()
+    
+    def _spawn_tile(self, value: Optional[int] = None) -> bool:
+        """
+        在空格生成新砖块
+        开局时value=2，后续随机2或4
+        返回是否成功生成
+        """
+        empty_cells = list(zip(*np.where(self.board == 0)))
+        if not empty_cells:
+            return False
+        
+        row, col = random.choice(empty_cells)
+        if value is None:
+            # 90%概率生成2，10%概率生成4
+            value = 2 if random.random() < 0.9 else 4
+        self.board[row, col] = value
+        return True
+    
+    def _compress(self, line: np.ndarray) -> Tuple[np.ndarray, int]:
+        """
+        压缩一行/列，将非零元素移到一端
+        返回压缩后的行和合并得分
+        """
+        # 移除零，填充到末尾
+        non_zero = line[line != 0]
+        new_line = np.zeros_like(line)
+        score = 0
+        
+        pos = 0
+        i = 0
+        while i < len(non_zero):
+            if i + 1 < len(non_zero) and non_zero[i] == non_zero[i + 1]:
+                # 合并
+                new_line[pos] = non_zero[i] * 2
+                score += new_line[pos]
+                i += 2
+            else:
+                new_line[pos] = non_zero[i]
+                i += 1
+            pos += 1
+        
+        return new_line, score
+    
+    def _move_left(self) -> Tuple[bool, int]:
+        """向左移动，返回(是否移动, 得分)"""
+        moved = False
+        total_score = 0
+        
+        for i in range(4):
+            original = self.board[i].copy()
+            new_line, score = self._compress(self.board[i])
+            self.board[i] = new_line
+            total_score += score
+            if not np.array_equal(original, new_line):
+                moved = True
+        
+        return moved, total_score
+    
+    def _move_right(self) -> Tuple[bool, int]:
+        """向右移动"""
+        moved = False
+        total_score = 0
+        
+        for i in range(4):
+            original = self.board[i].copy()
+            new_line, score = self._compress(self.board[i][::-1])
+            self.board[i] = new_line[::-1]
+            total_score += score
+            if not np.array_equal(original, self.board[i]):
+                moved = True
+        
+        return moved, total_score
+    
+    def _move_up(self) -> Tuple[bool, int]:
+        """向上移动"""
+        moved = False
+        total_score = 0
+        
+        for j in range(4):
+            original = self.board[:, j].copy()
+            new_line, score = self._compress(self.board[:, j])
+            self.board[:, j] = new_line
+            total_score += score
+            if not np.array_equal(original, new_line):
+                moved = True
+        
+        return moved, total_score
+    
+    def _move_down(self) -> Tuple[bool, int]:
+        """向下移动"""
+        moved = False
+        total_score = 0
+        
+        for j in range(4):
+            original = self.board[:, j].copy()
+            new_line, score = self._compress(self.board[:, j][::-1])
+            self.board[:, j] = new_line[::-1]
+            total_score += score
+            if not np.array_equal(original, self.board[:, j]):
+                moved = True
+        
+        return moved, total_score
+    
+    def move(self, direction: int) -> Tuple[np.ndarray, float, bool, bool]:
+        """
+        执行移动
+        
+        Args:
+            direction: 0=上, 1=下, 2=左, 3=右
+            
+        Returns:
+            state: 新状态
+            reward: 奖励（累积分数增量 + 局面分数变化）
+            moved: 是否成功移动
+            done: 游戏是否结束
+        """
+        if self.game_over:
+            return self.get_state(), 0.0, False, True
+        
+        old_accumulated = self.accumulated_score
+        old_situational = self.situational_score
+        
+        # 执行移动
+        if direction == self.UP:
+            moved, score = self._move_up()
+        elif direction == self.DOWN:
+            moved, score = self._move_down()
+        elif direction == self.LEFT:
+            moved, score = self._move_left()
+        elif direction == self.RIGHT:
+            moved, score = self._move_right()
+        else:
+            raise ValueError(f"Invalid direction: {direction}")
+        
+        if moved:
+            self.accumulated_score += score
+            self.moves_count += 1
+            self._spawn_tile()
+            self._update_situational_score()
+            
+            # 检查游戏是否结束
+            self.game_over = self._check_game_over()
+        
+        # 计算奖励
+        accumulated_delta = self.accumulated_score - old_accumulated
+        situational_delta = self.situational_score - old_situational
+        
+        # 奖励 = 局面分数变化 * 0.7 + 累积分数增量 * 0.3 / 100 (归一化)
+        reward = situational_delta * 0.7 + accumulated_delta * 0.003
+        
+        # 游戏结束惩罚
+        if self.game_over:
+            reward -= 10.0
+        
+        return self.get_state(), reward, moved, self.game_over
+    
+    def _check_game_over(self) -> bool:
+        """检查游戏是否结束"""
+        # 还有空格
+        if np.any(self.board == 0):
+            return False
+        
+        # 检查是否还能合并
+        for i in range(4):
+            for j in range(4):
+                if i < 3 and self.board[i, j] == self.board[i + 1, j]:
+                    return False
+                if j < 3 and self.board[i, j] == self.board[i, j + 1]:
+                    return False
+        
+        return True
+    
+    def _update_situational_score(self) -> None:
+        """
+        更新局面分数
+        局面分数 = 空格数 * 10 + 最大连续相邻数 * 15 + log2(最大数字) * 5 + 单调性奖励
+        """
+        empty_cells = np.sum(self.board == 0)
+        
+        # 计算最大连续相邻数字
+        max_consecutive = self._calculate_max_consecutive()
+        
+        # 最高数字的对数
+        max_tile = np.max(self.board)
+        max_tile_log = np.log2(max_tile) if max_tile > 0 else 0
+        
+        # 单调性评估（鼓励数字按方向排列）
+        monotonicity = self._calculate_monotonicity()
+        
+        # 局面分数
+        self.situational_score = (
+            empty_cells * 10 +
+            max_consecutive * 15 +
+            max_tile_log * 5 +
+            monotonicity * 5
+        )
+    
+    def _calculate_max_consecutive(self) -> int:
+        """
+        计算最大连续相邻数字数量
+        相邻砖块拥有相邻数字，如512 1024 2048为3
+        """
+        max_count = 0
+        
+        # 检查所有行
+        for i in range(4):
+            count = self._count_consecutive_in_line(self.board[i])
+            max_count = max(max_count, count)
+        
+        # 检查所有列
+        for j in range(4):
+            count = self._count_consecutive_in_line(self.board[:, j])
+            max_count = max(max_count, count)
+        
+        return max_count
+    
+    def _count_consecutive_in_line(self, line: np.ndarray) -> int:
+        """计算一行/列中的最大连续相邻数字"""
+        non_zero = line[line != 0]
+        if len(non_zero) < 2:
+            return 0
+        
+        max_count = 1
+        current_count = 1
+        
+        for i in range(1, len(non_zero)):
+            # 相邻数字：2的幂次相邻
+            if abs(np.log2(non_zero[i]) - np.log2(non_zero[i-1])) == 1:
+                current_count += 1
+                max_count = max(max_count, current_count)
+            else:
+                current_count = 1
+        
+        return max_count
+    
+    def _calculate_monotonicity(self) -> float:
+        """
+        计算单调性
+        鼓励数字在行/列上递增或递减
+        """
+        score = 0.0
+        
+        # 行单调性
+        for i in range(4):
+            row = self.board[i]
+            row = row[row != 0]
+            if len(row) >= 2:
+                # 检查递增
+                if all(row[i] <= row[i+1] for i in range(len(row)-1)):
+                    score += len(row) - 1
+                # 检查递减
+                elif all(row[i] >= row[i+1] for i in range(len(row)-1)):
+                    score += len(row) - 1
+        
+        # 列单调性
+        for j in range(4):
+            col = self.board[:, j]
+            col = col[col != 0]
+            if len(col) >= 2:
+                if all(col[i] <= col[i+1] for i in range(len(col)-1)):
+                    score += len(col) - 1
+                elif all(col[i] >= col[i+1] for i in range(len(col)-1)):
+                    score += len(col) - 1
+        
+        return score
+    
+    def get_state(self) -> np.ndarray:
+        """
+        获取当前状态表示
+        返回: (4, 4) 棋盘，值为log2(value)，空格为0
+        """
+        state = np.zeros((4, 4), dtype=np.float32)
+        non_zero_mask = self.board > 0
+        state[non_zero_mask] = np.log2(self.board[non_zero_mask])
+        return state
+    
+    def get_state_with_scores(self) -> np.ndarray:
+        """
+        获取带分数的状态表示
+        返回: (18,) 包含16个棋盘位置 + 2个分数
+        """
+        board_state = self.get_state().flatten()
+        
+        # 归一化分数
+        max_accumulated = 50000  # 假设最大累积分数
+        max_situational = 200    # 假设最大局面分数
+        
+        normalized_accumulated = min(self.accumulated_score / max_accumulated, 1.0)
+        normalized_situational = min(self.situational_score / max_situational, 1.0)
+        
+        return np.concatenate([
+            board_state / 15.0,  # 归一化到 [0, 1]，最大2048 = log2(2048) = 11
+            [normalized_accumulated, normalized_situational]
+        ]).astype(np.float32)
+    
+    def get_valid_actions(self) -> np.ndarray:
+        """获取当前可执行的动作"""
+        valid = np.zeros(4, dtype=bool)
+        
+        # 临时保存状态
+        old_board = self.board.copy()
+        old_accumulated = self.accumulated_score
+        
+        for direction in range(4):
+            if direction == self.UP:
+                moved, _ = self._move_up()
+            elif direction == self.DOWN:
+                moved, _ = self._move_down()
+            elif direction == self.LEFT:
+                moved, _ = self._move_left()
+            else:
+                moved, _ = self._move_right()
+            
+            valid[direction] = moved
+            self.board = old_board.copy()
+        
+        self.accumulated_score = old_accumulated
+        return valid
+    
+    def get_max_tile(self) -> int:
+        """获取最大砖块值"""
+        return int(np.max(self.board))
+    
+    def get_empty_cells_count(self) -> int:
+        """获取空格数量"""
+        return int(np.sum(self.board == 0))
+    
+    def __str__(self) -> str:
+        """字符串表示"""
+        result = []
+        for row in self.board:
+            result.append(" | ".join(f"{int(x):4d}" if x > 0 else "   ." for x in row))
+        return "\n".join(result)
+
+
+if __name__ == "__main__":
+    # 测试游戏
+    game = Game2048()
+    print("Initial state:")
+    print(game)
+    print(f"Accumulated score: {game.accumulated_score}")
+    print(f"Situational score: {game.situational_score}")
+    
+    # 测试一些移动
+    moves = ['UP', 'LEFT', 'DOWN', 'RIGHT']
+    for i in range(10):
+        direction = i % 4
+        state, reward, moved, done = game.move(direction)
+        print(f"\nMove {moves[direction]}: moved={moved}, done={done}")
+        print(game)
+        print(f"Reward: {reward:.2f}")

gui.py

@@ -0,0 +1,835 @@
+"""
+GUI界面 - PyQt5实现
+"""
+import sys
+import os
+import numpy as np
+from PyQt5.QtWidgets import (
+    QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
+    QGridLayout, QLabel, QPushButton, QComboBox, QGroupBox,
+    QFrame, QSplitter, QStatusBar, QProgressBar, QCheckBox,
+    QFileDialog, QMessageBox, QSpinBox
+)
+from PyQt5.QtCore import Qt, QTimer, QThread, pyqtSignal
+from PyQt5.QtGui import QFont, QColor, QPalette
+import matplotlib
+matplotlib.use('Qt5Agg')
+from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
+from matplotlib.figure import Figure
+from collections import deque
+import time
+import torch
+
+from game import Game2048
+from model import Game2048Transformer
+from trainer import PPOTrainer, RolloutBuffer, Transition
+
+
+# 砖块颜色配置
+TILE_COLORS = {
+    0: '#cdc1b4',
+    2: '#eee4da',
+    4: '#ede0c8',
+    8: '#f2b179',
+    16: '#f59563',
+    32: '#f67c5f',
+    64: '#f65e3b',
+    128: '#edcf72',
+    256: '#edcc61',
+    512: '#edc850',
+    1024: '#edc53f',
+    2048: '#edc22e',
+    4096: '#3c3a32',
+    8192: '#3c3a32',
+}
+
+TILE_TEXT_COLORS = {
+    2: '#776e65',
+    4: '#776e65',
+}
+
+
+class GameBoardWidget(QWidget):
+    """2048游戏面板组件"""
+    
+    def __init__(self, parent=None):
+        super().__init__(parent)
+        self.board = np.zeros((4, 4), dtype=np.int64)
+        self.cell_size = 80
+        self.padding = 5
+        self.init_ui()
+    
+    def init_ui(self):
+        """初始化UI"""
+        self.setFixedSize(
+            self.cell_size * 4 + self.padding * 5,
+            self.cell_size * 4 + self.padding * 5
+        )
+    
+    def set_board(self, board: np.ndarray):
+        """设置棋盘状态"""
+        self.board = board.copy()
+        self.update()
+    
+    def paintEvent(self, event):
+        """绘制棋盘"""
+        from PyQt5.QtGui import QPainter, QColor, QPen, QBrush, QFont
+        
+        painter = QPainter(self)
+        painter.setRenderHint(QPainter.Antialiasing)
+        
+        # 背景
+        painter.fillRect(self.rect(), QColor('#bbada0'))
+        
+        # 绘制每个格子
+        for i in range(4):
+            for j in range(4):
+                x = self.padding + j * (self.cell_size + self.padding)
+                y = self.padding + i * (self.cell_size + self.padding)
+                value = self.board[i, j]
+                
+                # 格子颜色
+                color = TILE_COLORS.get(value, '#3c3a32')
+                painter.fillRect(x, y, self.cell_size, self.cell_size, QColor(color))
+                
+                # 数字
+                if value > 0:
+                    # 文字颜色
+                    text_color = TILE_TEXT_COLORS.get(value, '#f9f6f2')
+                    painter.setPen(QColor(text_color))
+                    
+                    # 字体大小根据数字位数调整
+                    if value < 100:
+                        font_size = 32
+                    elif value < 1000:
+                        font_size = 28
+                    else:
+                        font_size = 22
+                    
+                    font = QFont('Arial', font_size, QFont.Bold)
+                    painter.setFont(font)
+                    
+                    # 居中绘制
+                    text = str(int(value))
+                    painter.drawText(x, y, self.cell_size, self.cell_size,
+                                    Qt.AlignCenter, text)
+
+
+class ScoreWidget(QWidget):
+    """分数显示组件"""
+    
+    def __init__(self, title: str, parent=None):
+        super().__init__(parent)
+        self.title = title
+        self.value = 0
+        self.init_ui()
+    
+    def init_ui(self):
+        """初始化UI"""
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(5, 5, 5, 5)
+        
+        # 标题
+        self.title_label = QLabel(self.title)
+        self.title_label.setAlignment(Qt.AlignCenter)
+        self.title_label.setStyleSheet("""
+            QLabel {
+                background-color: #bbada0;
+                color: #eee4da;
+                font-size: 12px;
+                font-weight: bold;
+                border-radius: 3px;
+                padding: 5px;
+            }
+        """)
+        layout.addWidget(self.title_label)
+        
+        # 数值
+        self.value_label = QLabel('0')
+        self.value_label.setAlignment(Qt.AlignCenter)
+        self.value_label.setStyleSheet("""
+            QLabel {
+                background-color: #8f7a66;
+                color: white;
+                font-size: 20px;
+                font-weight: bold;
+                border-radius: 3px;
+                padding: 10px;
+                min-width: 80px;
+            }
+        """)
+        layout.addWidget(self.value_label)
+    
+    def set_value(self, value):
+        """设置分数值"""
+        self.value = value
+        if isinstance(value, float):
+            self.value_label.setText(f'{value:.1f}')
+        else:
+            self.value_label.setText(str(int(value)))
+
+
+class PlotCanvas(FigureCanvas):
+    """matplotlib绑定的画布"""
+    
+    def __init__(self, parent=None, width=5, height=4, dpi=100):
+        self.fig = Figure(figsize=(width, height), dpi=dpi)
+        self.axes = self.fig.add_subplot(111)
+        super().__init__(self.fig)
+        self.setParent(parent)
+        
+        self.fig.patch.set_facecolor('#faf8ef')
+        self.axes.set_facecolor('#faf8ef')
+        
+        # 设置中文字体
+        self.axes.set_prop_cycle(color=['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728'])
+        
+        self.fig.tight_layout()
+    
+    def plot_training_scores(self, scores, title="Training Scores"):
+        """绘制训练曲线 - 只显示累积分数"""
+        self.axes.clear()
+        
+        if scores:
+            x = range(1, len(scores) + 1)
+            self.axes.plot(x, scores, '#1f77b4', linewidth=1.5, alpha=0.8)
+            
+            # 添加移动平均线
+            if len(scores) >= 10:
+                window = min(50, len(scores) // 5)
+                if window >= 5:
+                    ma = np.convolve(scores, np.ones(window)/window, mode='valid')
+                    ma_x = range(window, len(scores) + 1)
+                    self.axes.plot(ma_x, ma, '#d62728', linewidth=2, alpha=0.8, label=f'MA({window})')
+        
+        self.axes.set_xlabel('Games', fontsize=10)
+        self.axes.set_ylabel('Score', fontsize=10)
+        self.axes.set_title(title, fontsize=12)
+        self.axes.grid(True, alpha=0.3)
+        
+        self.fig.tight_layout()
+        self.draw()
+    
+    def plot_demo_scores(self, scores, title="Situational Score"):
+        """绘制演示曲线 - 只显示局面分数"""
+        self.axes.clear()
+        
+        if scores:
+            x = range(1, len(scores) + 1)
+            self.axes.plot(x, scores, '#2ca02c', linewidth=1.5, alpha=0.8)
+        
+        self.axes.set_xlabel('Steps', fontsize=10)
+        self.axes.set_ylabel('Situational Score', fontsize=10)
+        self.axes.set_title(title, fontsize=12)
+        self.axes.grid(True, alpha=0.3)
+        
+        self.fig.tight_layout()
+        self.draw()
+
+
+class SimpleTrainingThread(QThread):
+    """简化训练线程"""
+    
+    game_end_signal = pyqtSignal(dict)
+    progress_signal = pyqtSignal(dict)
+    finished_signal = pyqtSignal()
+    
+    def __init__(self, model, trainer, num_games=1000):
+        super().__init__()
+        self.model = model
+        self.trainer = trainer
+        self.num_games = num_games
+        self.running = True
+        self.device = "cpu"
+    
+    def run(self):
+        """运行训练"""
+        scores = []
+        max_tiles = []
+        best_score = 0
+        start_time = time.time()
+        
+        for game_idx in range(self.num_games):
+            if not self.running:
+                break
+            
+            game = Game2048()
+            game.reset()
+            buffer = RolloutBuffer(capacity=10000)
+            
+            while not game.game_over and self.running:
+                state = game.get_state()
+                scores_feat = np.array([
+                    min(game.accumulated_score / 50000, 1.0),
+                    min(game.situational_score / 200, 1.0)
+                ], dtype=np.float32)
+                valid = game.get_valid_actions()
+                
+                state_t = torch.FloatTensor(state).unsqueeze(0).to(self.device)
+                scores_t = torch.FloatTensor(scores_feat).unsqueeze(0).to(self.device)
+                valid_t = torch.BoolTensor(valid).unsqueeze(0).to(self.device)
+                
+                with torch.no_grad():
+                    action, log_prob, value = self.model.get_action(
+                        state_t, scores_t, valid_t, deterministic=False
+                    )
+                
+                old_state = state.copy()
+                old_scores = scores_feat.copy()
+                next_state, reward, moved, done = game.move(action)
+                
+                transition = Transition(
+                    state=old_state,
+                    scores=old_scores,
+                    action=action,
+                    reward=reward,
+                    next_state=next_state.copy(),
+                    next_scores=np.array([
+                        min(game.accumulated_score / 50000, 1.0),
+                        min(game.situational_score / 200, 1.0)
+                    ], dtype=np.float32),
+                    done=done,
+                    log_prob=log_prob,
+                    value=value,
+                    valid_actions=valid
+                )
+                buffer.push(transition)
+                
+                if len(buffer) >= 64:
+                    self.trainer.update(buffer)
+                    buffer.clear()
+            
+            scores.append(game.accumulated_score)
+            max_tiles.append(game.get_max_tile())
+            
+            if game.accumulated_score > best_score:
+                best_score = game.accumulated_score
+            
+            # 发送游戏结束信号
+            elapsed = time.time() - start_time
+            self.game_end_signal.emit({
+                'score': game.accumulated_score,
+                'max_tile': game.get_max_tile(),
+                'game_idx': game_idx + 1,
+                'best_score': best_score,
+                'elapsed': elapsed,
+                'avg_score': np.mean(scores[-100:]) if scores else 0
+            })
+        
+        self.finished_signal.emit()
+    
+    def stop(self):
+        """停止训练"""
+        self.running = False
+
+
+class MainWindow(QMainWindow):
+    """主窗口"""
+    
+    def __init__(self):
+        super().__init__()
+        self.setWindowTitle('2048 AI Trainer')
+        self.setMinimumSize(1000, 700)
+        
+        # 初始化模型和训练器
+        self.device = 'cpu'
+        self.model = Game2048Transformer()
+        self.trainer = PPOTrainer(self.model, lr=3e-4, device=self.device)
+        
+        # 游戏实例（用于演示）
+        self.game = Game2048()
+        
+        # 训练状态
+        self.is_training = False
+        self.training_thread = None
+        
+        # 统计数据
+        self.training_scores = []
+        self.demo_situational_scores = []
+        
+        # 演示模式状态
+        self.ai_mode = False
+        self.auto_step = False
+        self.auto_timer = QTimer()
+        self.auto_timer.timeout.connect(self.ai_step)
+        
+        # 当前模型路径
+        self.current_model_path = None
+        
+        self.init_ui()
+        self.update_display()
+    
+    def init_ui(self):
+        """初始化UI"""
+        # 主窗口样式
+        self.setStyleSheet("""
+            QMainWindow {
+                background-color: #faf8ef;
+            }
+            QLabel {
+                color: #776e65;
+            }
+            QPushButton {
+                background-color: #8f7a66;
+                color: white;
+                border: none;
+                border-radius: 3px;
+                padding: 10px 20px;
+                font-size: 14px;
+                font-weight: bold;
+                min-width: 80px;
+            }
+            QPushButton:hover {
+                background-color: #9f8b77;
+            }
+            QPushButton:pressed {
+                background-color: #7f6a57;
+            }
+            QPushButton:disabled {
+                background-color: #ccc;
+                color: #999;
+            }
+            QComboBox {
+                background-color: #8f7a66;
+                color: white;
+                border: none;
+                border-radius: 3px;
+                padding: 5px 10px;
+                min-width: 100px;
+            }
+            QComboBox::drop-down {
+                border: none;
+            }
+            QGroupBox {
+                border: 2px solid #bbada0;
+                border-radius: 5px;
+                margin-top: 10px;
+                padding-top: 10px;
+                font-weight: bold;
+                color: #776e65;
+            }
+            QGroupBox::title {
+                subcontrol-origin: margin;
+                left: 10px;
+                padding: 0 5px;
+            }
+            QSpinBox {
+                background-color: white;
+                border: 1px solid #bbada0;
+                border-radius: 3px;
+                padding: 5px;
+                min-width: 80px;
+            }
+        """)
+        
+        # 中央widget
+        central_widget = QWidget()
+        self.setCentralWidget(central_widget)
+        
+        # 主布局
+        main_layout = QHBoxLayout(central_widget)
+        
+        # 左侧面板（游戏区）
+        left_panel = self.create_left_panel()
+        main_layout.addWidget(left_panel)
+        
+        # 右侧面板（统计和控制）
+        right_panel = self.create_right_panel()
+        main_layout.addWidget(right_panel)
+        
+        # 设置拉伸比例
+        main_layout.setStretch(0, 1)
+        main_layout.setStretch(1, 2)
+        
+        # 状态栏
+        self.statusBar = QStatusBar()
+        self.setStatusBar(self.statusBar)
+        self.statusBar.showMessage('Ready')
+    
+    def create_left_panel(self) -> QWidget:
+        """创建左侧面板"""
+        panel = QWidget()
+        layout = QVBoxLayout(panel)
+        layout.setAlignment(Qt.AlignCenter)
+        
+        # 模式选择
+        mode_layout = QHBoxLayout()
+        mode_label = QLabel('Mode:')
+        mode_label.setFont(QFont('Arial', 12, QFont.Bold))
+        
+        self.mode_combo = QComboBox()
+        self.mode_combo.addItems(['Training Mode', 'Demo Mode'])
+        self.mode_combo.currentIndexChanged.connect(self.switch_mode)
+        
+        mode_layout.addWidget(mode_label)
+        mode_layout.addWidget(self.mode_combo)
+        mode_layout.addStretch()
+        layout.addLayout(mode_layout)
+        
+        # 分数显示
+        scores_layout = QHBoxLayout()
+        self.accumulated_score_widget = ScoreWidget('Score')
+        self.situational_score_widget = ScoreWidget('Situational')
+        self.max_tile_widget = ScoreWidget('Max Tile')
+        
+        scores_layout.addWidget(self.accumulated_score_widget)
+        scores_layout.addWidget(self.situational_score_widget)
+        scores_layout.addWidget(self.max_tile_widget)
+        layout.addLayout(scores_layout)
+        
+        # 游戏面板
+        self.game_board = GameBoardWidget()
+        layout.addWidget(self.game_board, alignment=Qt.AlignCenter)
+        
+        # 演示模式控制
+        self.demo_controls = QWidget()
+        demo_layout = QHBoxLayout(self.demo_controls)
+        
+        self.ai_btn = QPushButton('AI Mode')
+        self.ai_btn.clicked.connect(self.toggle_ai_mode)
+        
+        self.step_btn = QPushButton('Step')
+        self.step_btn.clicked.connect(self.ai_step)
+        
+        self.auto_btn = QPushButton('Auto')
+        self.auto_btn.clicked.connect(self.toggle_auto)
+        
+        self.reset_btn = QPushButton('Reset')
+        self.reset_btn.clicked.connect(self.reset_game)
+        
+        demo_layout.addWidget(self.ai_btn)
+        demo_layout.addWidget(self.step_btn)
+        demo_layout.addWidget(self.auto_btn)
+        demo_layout.addWidget(self.reset_btn)
+        
+        self.demo_controls.setVisible(False)
+        layout.addWidget(self.demo_controls)
+        
+        layout.addStretch()
+        return panel
+    
+    def create_right_panel(self) -> QWidget:
+        """创建右侧面板"""
+        panel = QWidget()
+        layout = QVBoxLayout(panel)
+        
+        # 训练控制
+        control_group = QGroupBox('Training Control')
+        control_layout = QVBoxLayout(control_group)
+        
+        # 训练局数设置
+        games_layout = QHBoxLayout()
+        games_label = QLabel('Games:')
+        self.games_spinbox = QSpinBox()
+        self.games_spinbox.setRange(10, 100000)
+        self.games_spinbox.setValue(500)
+        games_layout.addWidget(games_label)
+        games_layout.addWidget(self.games_spinbox)
+        games_layout.addStretch()
+        control_layout.addLayout(games_layout)
+        
+        # 按钮
+        btn_layout = QHBoxLayout()
+        self.start_btn = QPushButton('Start Training')
+        self.start_btn.clicked.connect(self.start_training)
+        
+        self.stop_btn = QPushButton('Stop Training')
+        self.stop_btn.clicked.connect(self.stop_training)
+        self.stop_btn.setEnabled(False)
+        
+        btn_layout.addWidget(self.start_btn)
+        btn_layout.addWidget(self.stop_btn)
+        control_layout.addLayout(btn_layout)
+        
+        # 模型文件操作
+        model_layout = QHBoxLayout()
+        self.load_btn = QPushButton('Load Model')
+        self.load_btn.clicked.connect(self.load_model)
+        
+        self.save_btn = QPushButton('Save Model')
+        self.save_btn.clicked.connect(self.save_model)
+        
+        model_layout.addWidget(self.load_btn)
+        model_layout.addWidget(self.save_btn)
+        control_layout.addLayout(model_layout)
+        
+        # 训练参数显示
+        param_layout = QGridLayout()
+        
+        self.games_label = QLabel('Games: 0')
+        self.avg_score_label = QLabel('Avg Score: 0')
+        self.best_score_label = QLabel('Best Score: 0')
+        self.speed_label = QLabel('Speed: 0 games/s')
+        
+        param_layout.addWidget(self.games_label, 0, 0)
+        param_layout.addWidget(self.avg_score_label, 0, 1)
+        param_layout.addWidget(self.best_score_label, 1, 0)
+        param_layout.addWidget(self.speed_label, 1, 1)
+        
+        control_layout.addLayout(param_layout)
+        layout.addWidget(control_group)
+        
+        # 分数曲线
+        plot_group = QGroupBox('Score Chart')
+        plot_layout = QVBoxLayout(plot_group)
+        
+        self.plot_canvas = PlotCanvas(self, width=6, height=4, dpi=100)
+        plot_layout.addWidget(self.plot_canvas)
+        
+        layout.addWidget(plot_group)
+        
+        # 实时统计
+        stats_group = QGroupBox('Training Stats')
+        stats_layout = QVBoxLayout(stats_group)
+        
+        self.stats_text = QLabel('Waiting for training...')
+        self.stats_text.setStyleSheet('font-family: monospace;')
+        stats_layout.addWidget(self.stats_text)
+        
+        layout.addWidget(stats_group)
+        
+        return panel
+    
+    def switch_mode(self, index):
+        """切换模式"""
+        if index == 0:  # 训练模式
+            self.demo_controls.setVisible(False)
+            self.demo_situational_scores = []
+            self.plot_canvas.plot_training_scores(self.training_scores)
+        else:  # 演示模式
+            self.demo_controls.setVisible(True)
+            self.reset_game()
+            self.demo_situational_scores = []
+            self.plot_canvas.plot_demo_scores([])
+    
+    def keyPressEvent(self, event):
+        """键盘事件"""
+        if self.mode_combo.currentIndex() == 1 and not self.ai_mode:
+            # 演示模式且非AI托管
+            key_map = {
+                Qt.Key_Up: 0,
+                Qt.Key_Down: 1,
+                Qt.Key_Left: 2,
+                Qt.Key_Right: 3,
+            }
+            
+            if event.key() in key_map:
+                direction = key_map[event.key()]
+                self.game.move(direction)
+                self.demo_situational_scores.append(self.game.situational_score)
+                self.update_display()
+                self.plot_canvas.plot_demo_scores(self.demo_situational_scores)
+                
+                if self.game.game_over:
+                    self.statusBar.showMessage('Game Over!')
+    
+    def toggle_ai_mode(self):
+        """切换AI托管模式"""
+        self.ai_mode = not self.ai_mode
+        if self.ai_mode:
+            self.ai_btn.setText('Manual')
+            self.step_btn.setEnabled(False)
+            self.auto_btn.setEnabled(True)
+        else:
+            self.ai_btn.setText('AI Mode')
+            self.step_btn.setEnabled(True)
+            self.auto_btn.setEnabled(False)
+            self.auto_timer.stop()
+            self.auto_btn.setText('Auto')
+    
+    def ai_step(self):
+        """AI单步执行"""
+        if self.game.game_over:
+            self.reset_game()
+            return
+        
+        state = self.game.get_state()
+        scores = np.array([
+            min(self.game.accumulated_score / 50000, 1.0),
+            min(self.game.situational_score / 200, 1.0)
+        ], dtype=np.float32)
+        valid_actions = self.game.get_valid_actions()
+        
+        state_t = torch.FloatTensor(state).unsqueeze(0)
+        scores_t = torch.FloatTensor(scores).unsqueeze(0)
+        valid_t = torch.BoolTensor(valid_actions).unsqueeze(0)
+        
+        action, _, _ = self.model.get_action(state_t, scores_t, valid_t, deterministic=True)
+        
+        self.game.move(action)
+        self.demo_situational_scores.append(self.game.situational_score)
+        self.update_display()
+        self.plot_canvas.plot_demo_scores(self.demo_situational_scores)
+        
+        if self.game.game_over:
+            self.statusBar.showMessage(f'Game Over! Final Score: {self.game.accumulated_score}')
+            if self.auto_timer.isActive():
+                self.auto_timer.stop()
+                self.auto_btn.setText('Auto')
+    
+    def toggle_auto(self):
+        """切换自动执行"""
+        if self.auto_timer.isActive():
+            self.auto_timer.stop()
+            self.auto_btn.setText('Auto')
+        else:
+            self.auto_timer.start(100)  # 100ms间隔
+            self.auto_btn.setText('Stop')
+    
+    def reset_game(self):
+        """重置游戏"""
+        self.game.reset()
+        self.demo_situational_scores = [self.game.situational_score]
+        self.update_display()
+        self.plot_canvas.plot_demo_scores(self.demo_situational_scores)
+        self.statusBar.showMessage('Game Reset')
+    
+    def update_display(self):
+        """更新显示"""
+        self.game_board.set_board(self.game.board)
+        self.accumulated_score_widget.set_value(self.game.accumulated_score)
+        self.situational_score_widget.set_value(self.game.situational_score)
+        self.max_tile_widget.set_value(self.game.get_max_tile())
+    
+    def start_training(self):
+        """开始训练"""
+        self.is_training = True
+        self.start_btn.setEnabled(False)
+        self.stop_btn.setEnabled(True)
+        self.mode_combo.setEnabled(False)
+        self.games_spinbox.setEnabled(False)
+        self.load_btn.setEnabled(False)
+        self.save_btn.setEnabled(False)
+        
+        # 重置统计
+        self.training_scores = []
+        
+        # 创建训练线程
+        num_games = self.games_spinbox.value()
+        self.training_thread = SimpleTrainingThread(
+            self.model, self.trainer, num_games
+        )
+        self.training_thread.game_end_signal.connect(self.on_game_end)
+        self.training_thread.finished_signal.connect(self.on_training_finished)
+        self.training_thread.start()
+        
+        self.statusBar.showMessage('Training started...')
+    
+    def stop_training(self):
+        """停止训练"""
+        if self.training_thread:
+            self.training_thread.stop()
+            self.training_thread.wait()
+            self.training_thread = None
+        
+        self.on_training_finished()
+    
+    def on_game_end(self, stats):
+        """游戏结束回调"""
+        self.training_scores.append(stats['score'])
+        
+        # 更新统计显示
+        games = stats['game_idx']
+        avg_score = stats['avg_score']
+        best_score = stats['best_score']
+        elapsed = stats['elapsed']
+        speed = games / elapsed if elapsed > 0 else 0
+        
+        self.games_label.setText(f'Games: {games}')
+        self.avg_score_label.setText(f'Avg Score: {avg_score:.0f}')
+        self.best_score_label.setText(f'Best Score: {best_score}')
+        self.speed_label.setText(f'Speed: {speed:.2f} games/s')
+        
+        # 更新曲线
+        if games % 5 == 0:
+            self.plot_canvas.plot_training_scores(self.training_scores)
+        
+        # 更新统计文本
+        self.stats_text.setText(
+            f"Games: {games}\n"
+            f"Avg Score: {avg_score:.0f}\n"
+            f"Best Score: {best_score}\n"
+            f"Last Score: {stats['score']}"
+        )
+    
+    def on_training_finished(self):
+        """训练完成回调"""
+        self.is_training = False
+        self.start_btn.setEnabled(True)
+        self.stop_btn.setEnabled(False)
+        self.mode_combo.setEnabled(True)
+        self.games_spinbox.setEnabled(True)
+        self.load_btn.setEnabled(True)
+        self.save_btn.setEnabled(True)
+        
+        # 自动保存模型
+        save_path = os.path.join(os.path.dirname(__file__), 'checkpoints', 'model.pt')
+        os.makedirs(os.path.dirname(save_path), exist_ok=True)
+        torch.save({
+            'model_state_dict': self.model.state_dict(),
+            'training_scores': self.training_scores,
+            'best_score': max(self.training_scores) if self.training_scores else 0
+        }, save_path)
+        
+        self.statusBar.showMessage(f'Training finished! Model saved to {save_path}')
+    
+    def load_model(self):
+        """加载模型"""
+        file_path, _ = QFileDialog.getOpenFileName(
+            self, 'Load Model', 
+            os.path.join(os.path.dirname(__file__), 'checkpoints'),
+            'PyTorch Model (*.pt);;All Files (*)'
+        )
+        
+        if file_path:
+            try:
+                checkpoint = torch.load(file_path, map_location=self.device, weights_only=False)
+                self.model.load_state_dict(checkpoint['model_state_dict'])
+                self.current_model_path = file_path
+                
+                if 'training_scores' in checkpoint:
+                    self.training_scores = checkpoint['training_scores']
+                    self.plot_canvas.plot_training_scores(self.training_scores)
+                
+                self.statusBar.showMessage(f'Model loaded: {os.path.basename(file_path)}')
+            except Exception as e:
+                QMessageBox.warning(self, 'Error', f'Failed to load model:\n{str(e)}')
+    
+    def save_model(self):
+        """保存模型"""
+        file_path, _ = QFileDialog.getSaveFileName(
+            self, 'Save Model',
+            os.path.join(os.path.dirname(__file__), 'checkpoints', 'model.pt'),
+            'PyTorch Model (*.pt);;All Files (*)'
+        )
+        
+        if file_path:
+            try:
+                torch.save({
+                    'model_state_dict': self.model.state_dict(),
+                    'training_scores': self.training_scores,
+                    'best_score': max(self.training_scores) if self.training_scores else 0
+                }, file_path)
+                self.current_model_path = file_path
+                self.statusBar.showMessage(f'Model saved: {os.path.basename(file_path)}')
+            except Exception as e:
+                QMessageBox.warning(self, 'Error', f'Failed to save model:\n{str(e)}')
+
+
+def main():
+    """主函数"""
+    app = QApplication(sys.argv)
+    app.setStyle('Fusion')
+    
+    # 设置字体
+    font = QFont('Arial', 10)
+    app.setFont(font)
+    
+    window = MainWindow()
+    window.show()
+    
+    sys.exit(app.exec_())
+
+
+if __name__ == '__main__':
+    main()

main.py

@@ -0,0 +1,302 @@
+"""
+2048 AI Trainer - 主入口
+基于Transformer的2048游戏AI训练器
+
+使用方法:
+    python main.py              # 启动GUI界面
+    python main.py --train      # 命令行训练模式
+    python main.py --demo       # 演示模式（加载已有模型）
+"""
+import sys
+import os
+import argparse
+import torch
+import time
+
+# 添加当前目录到路径
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from game import Game2048
+from model import Game2048Transformer, count_parameters
+from trainer import PPOTrainer, RolloutBuffer, TrainingStats
+from parallel import TrainingLoop, ParallelGameEnv, TrainingWorker
+from utils import (
+    set_seed, get_device, print_model_info, save_checkpoint,
+    load_checkpoint, EarlyStopping, format_time, format_number
+)
+
+
+def run_gui():
+    """运行GUI界面"""
+    from gui import main
+    main()
+
+
+def run_training(
+    num_games: int = 10000,
+    num_envs: int = 4,
+    save_interval: int = 100,
+    checkpoint_dir: str = "checkpoints",
+    seed: int = 42
+):
+    """
+    命令行训练模式
+    
+    Args:
+        num_games: 总游戏局数
+        num_envs: 并行环境数
+        save_interval: 保存间隔
+        checkpoint_dir: 检查点目录
+        seed: 随机种子
+    """
+    print("=" * 50)
+    print("2048 AI Training")
+    print("=" * 50)
+    
+    # 设置随机种子
+    set_seed(seed)
+    
+    # 获取设备
+    device = get_device()
+    print(f"Device: {device}")
+    
+    # 创建模型
+    model = Game2048Transformer()
+    print_model_info(model)
+    model.to(device)
+    
+    # 创建训练器
+    trainer = PPOTrainer(model, lr=1e-4, device=device)
+    
+    # 创建训练循环
+    training_loop = TrainingLoop(
+        model=model,
+        trainer=trainer,
+        num_envs=num_envs,
+        device=device,
+        steps_per_update=256,
+        checkpoint_dir=checkpoint_dir
+    )
+    
+    # 训练统计
+    stats = TrainingStats()
+    start_time = time.time()
+    best_score = 0
+    
+    # 创建检查点目录
+    os.makedirs(checkpoint_dir, exist_ok=True)
+    
+    print(f"\nStarting training for {num_games} games...")
+    print(f"Parallel environments: {num_envs}")
+    print("-" * 50)
+    
+    try:
+        games_completed = 0
+        
+        def on_game_end(game_stats):
+            nonlocal games_completed, best_score
+            
+            stats.record_game(
+                score=game_stats['score'],
+                situational_score=game_stats['situational_score'],
+                max_tile=game_stats['max_tile'],
+                steps=game_stats['moves']
+            )
+            
+            games_completed += 1
+            
+            if game_stats['score'] > best_score:
+                best_score = game_stats['score']
+            
+            # 定期打印统计
+            if games_completed % 10 == 0:
+                elapsed = time.time() - start_time
+                avg_stats = stats.get_avg_stats(window=100)
+                
+                print(
+                    f"Games: {games_completed} | "
+                    f"Avg Score: {avg_stats['avg_score']:.0f} | "
+                    f"Best: {best_score} | "
+                    f"Max Tile: {avg_stats['avg_max_tile']:.0f} | "
+                    f"Speed: {games_completed/elapsed:.2f} games/s"
+                )
+            
+            # 保存检查点
+            if games_completed % save_interval == 0:
+                checkpoint_path = os.path.join(
+                    checkpoint_dir, 
+                    f"checkpoint_{games_completed}.pt"
+                )
+                save_checkpoint(
+                    model, trainer.optimizer, games_completed,
+                    avg_stats, checkpoint_path
+                )
+                print(f"Checkpoint saved: {checkpoint_path}")
+        
+        training_loop.on_game_end_callback = on_game_end
+        training_loop.train(total_games=num_games, stop_threshold=200)
+        
+    except KeyboardInterrupt:
+        print("\nTraining interrupted by user.")
+    
+    # 训练结束统计
+    elapsed = time.time() - start_time
+    final_stats = stats.get_avg_stats()
+    
+    print("\n" + "=" * 50)
+    print("Training Complete!")
+    print("=" * 50)
+    print(f"Total games: {format_number(final_stats['games_played'])}")
+    print(f"Total time: {format_time(elapsed)}")
+    print(f"Average score: {final_stats['avg_score']:.0f}")
+    print(f"Best score: {final_stats['best_score']}")
+    print(f"Best max tile: {final_stats['best_max_tile']}")
+    
+    # 保存最终模型
+    final_path = os.path.join(checkpoint_dir, "final_model.pt")
+    save_checkpoint(
+        model, trainer.optimizer, final_stats['games_played'],
+        final_stats, final_path
+    )
+    print(f"Final model saved: {final_path}")
+
+
+def run_demo(model_path: str = None, num_games: int = 5):
+    """
+    演示模式
+    
+    Args:
+        model_path: 模型路径
+        num_games: 演示游戏数
+    """
+    print("=" * 50)
+    print("2048 AI Demo Mode")
+    print("=" * 50)
+    
+    device = get_device()
+    print(f"Device: {device}")
+    
+    # 创建模型
+    model = Game2048Transformer()
+    
+    # 加载模型
+    if model_path and os.path.exists(model_path):
+        print(f"Loading model from: {model_path}")
+        checkpoint = load_checkpoint(model_path, model, device=device)
+        print(f"Loaded checkpoint: {checkpoint.get('epoch', 'unknown')} games")
+    else:
+        print("No model loaded, using random weights.")
+    
+    model.to(device)
+    model.eval()
+    print_model_info(model)
+    
+    # 运行演示游戏
+    print(f"\nRunning {num_games} demo games...")
+    print("-" * 50)
+    
+    total_scores = []
+    
+    for game_idx in range(num_games):
+        game = Game2048()
+        game.reset()
+        
+        steps = 0
+        while not game.game_over and steps < 10000:
+            # 获取状态
+            state = game.get_state()
+            scores = game.get_state_with_scores()[-2:]
+            valid_actions = game.get_valid_actions()
+            
+            # 转换为张量
+            state_t = torch.FloatTensor(state).unsqueeze(0).to(device)
+            scores_t = torch.FloatTensor(scores).unsqueeze(0).to(device)
+            valid_t = torch.BoolTensor(valid_actions).unsqueeze(0).to(device)
+            
+            # 获取动作
+            with torch.no_grad():
+                action, _, _ = model.get_action(state_t, scores_t, valid_t, deterministic=True)
+            
+            # 执行动作
+            game.move(action)
+            steps += 1
+        
+        total_scores.append(game.accumulated_score)
+        
+        print(
+            f"Game {game_idx + 1}: "
+            f"Score = {game.accumulated_score}, "
+            f"Max Tile = {game.get_max_tile()}, "
+            f"Steps = {steps}"
+        )
+    
+    # 统计
+    print("\n" + "-" * 50)
+    print(f"Average score: {sum(total_scores) / len(total_scores):.0f}")
+    print(f"Best score: {max(total_scores)}")
+    print(f"Worst score: {min(total_scores)}")
+
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(
+        description='2048 AI Trainer - 基于Transformer的2048游戏AI'
+    )
+    
+    parser.add_argument(
+        '--train', 
+        action='store_true',
+        help='运行命令行训练模式'
+    )
+    
+    parser.add_argument(
+        '--demo',
+        action='store_true',
+        help='运行演示模式'
+    )
+    
+    parser.add_argument(
+        '--model',
+        type=str,
+        default=None,
+        help='模型路径（用于演示模式或继续训练）'
+    )
+    
+    parser.add_argument(
+        '--games',
+        type=int,
+        default=10000,
+        help='训练游戏数（默认10000）'
+    )
+    
+    parser.add_argument(
+        '--envs',
+        type=int,
+        default=4,
+        help='并行环境数（默认4）'
+    )
+    
+    parser.add_argument(
+        '--seed',
+        type=int,
+        default=42,
+        help='随机种子（默认42）'
+    )
+    
+    args = parser.parse_args()
+    
+    if args.train:
+        run_training(
+            num_games=args.games,
+            num_envs=args.envs,
+            seed=args.seed
+        )
+    elif args.demo:
+        run_demo(model_path=args.model, num_games=5)
+    else:
+        # 默认启动GUI
+        run_gui()
+
+
+if __name__ == "__main__":
+    main()

model.py

@@ -0,0 +1,341 @@
+"""
+Transformer模型用于2048游戏决策
+"""
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import numpy as np
+from typing import Tuple
+
+
+class Game2048Transformer(nn.Module):
+    """
+    小型Transformer模型用于2048游戏
+    
+    输入: 
+        - 棋盘状态 (batch, 4, 4) 或 (batch, 16) 
+        - 可选: 分数特征 (batch, 2)
+    
+    输出:
+        - policy: (batch, 4) 动作概率
+        - value: (batch, 1) 状态价值
+    """
+    
+    def __init__(
+        self,
+        d_model: int = 64,
+        nhead: int = 4,
+        num_layers: int = 2,
+        dim_feedforward: int = 128,
+        dropout: float = 0.1
+    ):
+        super().__init__()
+        
+        self.d_model = d_model
+        
+        # 棋盘位置embedding
+        # 每个格子: 0表示空，1-15表示log2(value)
+        self.position_embedding = nn.Embedding(16, d_model)
+        
+        # 空间位置编码（4x4棋盘的行列位置）
+        self.row_embedding = nn.Embedding(4, d_model // 2)
+        self.col_embedding = nn.Embedding(4, d_model // 2)
+        
+        # 分数特征embedding
+        self.score_embedding = nn.Linear(2, d_model)
+        
+        # Transformer编码器
+        encoder_layer = nn.TransformerEncoderLayer(
+            d_model=d_model,
+            nhead=nhead,
+            dim_feedforward=dim_feedforward,
+            dropout=dropout,
+            batch_first=True,
+            activation='gelu'
+        )
+        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
+        
+        # 策略头（输出4个动作的概率）
+        self.policy_head = nn.Sequential(
+            nn.Linear(d_model, d_model),
+            nn.GELU(),
+            nn.Linear(d_model, 4)
+        )
+        
+        # 价值头（输出状态价值）
+        self.value_head = nn.Sequential(
+            nn.Linear(d_model, d_model),
+            nn.GELU(),
+            nn.Linear(d_model, 1),
+            nn.Tanh()
+        )
+        
+        # 初始化权重
+        self._init_weights()
+    
+    def _init_weights(self):
+        """初始化权重"""
+        for module in self.modules():
+            if isinstance(module, nn.Linear):
+                nn.init.xavier_uniform_(module.weight)
+                if module.bias is not None:
+                    nn.init.zeros_(module.bias)
+            elif isinstance(module, nn.Embedding):
+                nn.init.normal_(module.weight, mean=0, std=0.02)
+    
+    def forward(
+        self, 
+        board: torch.Tensor, 
+        scores: torch.Tensor = None,
+        mask: torch.Tensor = None
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
+        """
+        前向传播
+        
+        Args:
+            board: (batch, 4, 4) 或 (batch, 16) 棋盘状态，值为log2(value)
+            scores: (batch, 2) 可选的分数特征 [累积分数, 局面分数]
+            mask: (batch, 4) 可选的动作mask，True表示有效动作
+            
+        Returns:
+            policy: (batch, 4) 动作logits
+            value: (batch, 1) 状态价值
+        """
+        batch_size = board.shape[0]
+        
+        # 展平棋盘
+        if board.dim() == 3:
+            board_flat = board.view(batch_size, -1)  # (batch, 16)
+        else:
+            board_flat = board
+        
+        # 将棋盘值转换为embedding索引（clamp到有效范围）
+        board_indices = torch.clamp(board_flat.long(), 0, 15)
+        
+        # 位置embedding
+        pos_embeddings = self.position_embedding(board_indices)  # (batch, 16, d_model)
+        
+        # 添加空间位置编码
+        row_indices = torch.arange(4, device=board.device).repeat(4)
+        col_indices = torch.arange(4, device=board.device).repeat_interleave(4)
+        row_emb = self.row_embedding(row_indices)  # (16, d_model//2)
+        col_emb = self.col_embedding(col_indices)  # (16, d_model//2)
+        spatial_emb = torch.cat([row_emb, col_emb], dim=-1)  # (16, d_model)
+        
+        # 合并embedding
+        x = pos_embeddings + spatial_emb.unsqueeze(0)  # (batch, 16, d_model)
+        
+        # 如果提供分数特征，作为第17个token
+        if scores is not None:
+            score_emb = self.score_embedding(scores).unsqueeze(1)  # (batch, 1, d_model)
+            x = torch.cat([x, score_emb], dim=1)  # (batch, 17, d_model)
+        
+        # Transformer编码
+        x = self.transformer(x)  # (batch, 17 or 16, d_model)
+        
+        # 全局池化
+        x = x.mean(dim=1)  # (batch, d_model)
+        
+        # 输出头
+        policy_logits = self.policy_head(x)  # (batch, 4)
+        value = self.value_head(x)  # (batch, 1)
+        
+        # 应用动作mask
+        if mask is not None:
+            # 无效动作设为很小的值
+            policy_logits = policy_logits.masked_fill(~mask, -1e9)
+        
+        return policy_logits, value
+    
+    def get_action(
+        self, 
+        board: torch.Tensor, 
+        scores: torch.Tensor = None,
+        mask: torch.Tensor = None,
+        deterministic: bool = False
+    ) -> Tuple[int, torch.Tensor, torch.Tensor]:
+        """
+        选择动作
+        
+        Args:
+            board: (1, 4, 4) 或 (4, 4) 棋盘状态
+            scores: (1, 2) 或 (2,) 分数特征
+            mask: (1, 4) 或 (4,) 动作mask
+            deterministic: 是否确定性选择
+            
+        Returns:
+            action: 选择的动作
+            log_prob: 动作的log概率
+            value: 状态价值
+        """
+        # 确保维度正确
+        if board.dim() == 2:
+            board = board.unsqueeze(0)
+        if scores is not None and scores.dim() == 1:
+            scores = scores.unsqueeze(0)
+        if mask is not None and mask.dim() == 1:
+            mask = mask.unsqueeze(0)
+        
+        with torch.no_grad():
+            policy_logits, value = self.forward(board, scores, mask)
+            probs = F.softmax(policy_logits, dim=-1)
+            
+            if deterministic:
+                action = torch.argmax(probs, dim=-1).item()
+            else:
+                # 从概率分布采样
+                dist = torch.distributions.Categorical(probs)
+                action = dist.sample().item()
+            
+            log_prob = F.log_softmax(policy_logits, dim=-1)[0, action].item()
+        
+        return action, log_prob, value.item()
+    
+    def evaluate_actions(
+        self,
+        board: torch.Tensor,
+        actions: torch.Tensor,
+        scores: torch.Tensor = None,
+        mask: torch.Tensor = None
+    ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
+        """
+        评估动作（用于训练）
+        
+        Args:
+            board: (batch, 4, 4) 棋盘状态
+            actions: (batch,) 采取的动作
+            scores: (batch, 2) 分数特征
+            mask: (batch, 4) 动作mask
+            
+        Returns:
+            log_probs: (batch,) 动作log概率
+            values: (batch, 1) 状态价值
+            entropy: (batch,) 策略熵
+        """
+        policy_logits, values = self.forward(board, scores, mask)
+        
+        probs = F.softmax(policy_logits, dim=-1)
+        log_probs = F.log_softmax(policy_logits, dim=-1)
+        
+        # 选择动作的log概率
+        action_log_probs = log_probs.gather(1, actions.unsqueeze(1)).squeeze(1)
+        
+        # 计算熵
+        entropy = -(probs * log_probs).sum(dim=-1)
+        
+        return action_log_probs, values, entropy
+
+
+class ResidualBlock(nn.Module):
+    """残差块"""
+    
+    def __init__(self, channels: int):
+        super().__init__()
+        self.conv1 = nn.Conv2d(channels, channels, kernel_size=3, padding=1)
+        self.conv2 = nn.Conv2d(channels, channels, kernel_size=3, padding=1)
+        self.bn1 = nn.BatchNorm2d(channels)
+        self.bn2 = nn.BatchNorm2d(channels)
+    
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        residual = x
+        x = F.relu(self.bn1(self.conv1(x)))
+        x = self.bn2(self.conv2(x))
+        return F.relu(x + residual)
+
+
+class Game2048CNN(nn.Module):
+    """
+    CNN版本的2048模型（作为备选）
+    更简单，可能更快
+    """
+    
+    def __init__(self, channels: int = 64):
+        super().__init__()
+        
+        # 输入: (batch, 1, 4, 4)
+        self.conv1 = nn.Conv2d(1, channels, kernel_size=3, padding=1)
+        self.conv2 = nn.Conv2d(channels, channels, kernel_size=3, padding=1)
+        self.conv3 = nn.Conv2d(channels, channels, kernel_size=3, padding=1)
+        
+        # 分数处理
+        self.score_fc = nn.Linear(2, channels)
+        
+        # 输出头
+        self.policy_head = nn.Sequential(
+            nn.Linear(channels * 16 + channels, 256),
+            nn.ReLU(),
+            nn.Linear(256, 4)
+        )
+        
+        self.value_head = nn.Sequential(
+            nn.Linear(channels * 16 + channels, 256),
+            nn.ReLU(),
+            nn.Linear(256, 1),
+            nn.Tanh()
+        )
+    
+    def forward(
+        self, 
+        board: torch.Tensor, 
+        scores: torch.Tensor = None,
+        mask: torch.Tensor = None
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
+        """前向传播"""
+        # 添加channel维度
+        x = board.unsqueeze(1)  # (batch, 1, 4, 4)
+        
+        # CNN特征提取
+        x = F.relu(self.conv1(x))
+        x = F.relu(self.conv2(x))
+        x = F.relu(self.conv3(x))
+        
+        # 展平
+        x = x.view(x.size(0), -1)  # (batch, channels*16)
+        
+        # 合并分数
+        if scores is not None:
+            score_feat = F.relu(self.score_fc(scores))
+            x = torch.cat([x, score_feat], dim=-1)
+        else:
+            x = torch.cat([x, torch.zeros(x.size(0), 64, device=x.device)], dim=-1)
+        
+        # 输出
+        policy_logits = self.policy_head(x)
+        value = self.value_head(x)
+        
+        if mask is not None:
+            policy_logits = policy_logits.masked_fill(~mask, -1e9)
+        
+        return policy_logits, value
+
+
+def count_parameters(model: nn.Module) -> int:
+    """计算模型参数数量"""
+    return sum(p.numel() for p in model.parameters() if p.requires_grad)
+
+
+if __name__ == "__main__":
+    # 测试模型
+    device = torch.device("cpu")
+    
+    # Transformer模型
+    model = Game2048Transformer().to(device)
+    print(f"Transformer参数量: {count_parameters(model):,}")
+    
+    # 测试前向传播
+    batch_size = 4
+    board = torch.randint(0, 12, (batch_size, 4, 4), dtype=torch.float32).to(device)
+    scores = torch.rand(batch_size, 2).to(device)
+    mask = torch.ones(batch_size, 4, dtype=torch.bool).to(device)
+    
+    policy_logits, value = model(board, scores, mask)
+    print(f"Policy logits shape: {policy_logits.shape}")
+    print(f"Value shape: {value.shape}")
+    
+    # 测试动作选择
+    action, log_prob, val = model.get_action(board[0], scores[0], mask[0])
+    print(f"Action: {action}, Log prob: {log_prob:.4f}, Value: {val:.4f}")
+    
+    # CNN模型
+    cnn_model = Game2048CNN().to(device)
+    print(f"\nCNN参数量: {count_parameters(cnn_model):,}")

parallel.py

@@ -0,0 +1,461 @@
+"""
+多进程并行训练环境
+"""
+import numpy as np
+import torch
+from typing import List, Tuple, Optional
+from concurrent.futures import ThreadPoolExecutor
+import threading
+from game import Game2048
+from trainer import Transition, TrainingStats
+import time
+
+
+class ParallelGameEnv:
+    """
+    并行游戏环境
+    使用多线程同时运行多个游戏实例
+    """
+    
+    def __init__(self, num_envs: int = 4):
+        """
+        初始化并行环境
+        
+        Args:
+            num_envs: 并行游戏数量
+        """
+        self.num_envs = num_envs
+        self.envs = [Game2048() for _ in range(num_envs)]
+        self.states = [env.get_state() for env in self.envs]
+        self.scores = [env.accumulated_score for env in self.envs]
+        self.situational_scores = [env.situational_score for env in self.envs]
+        
+        # 游戏结束回调
+        self.on_game_end = None
+        
+        # 已完成游戏计数
+        self.games_completed = 0
+        
+        # 使用线程池
+        self.executor = ThreadPoolExecutor(max_workers=num_envs)
+    
+    def reset(self, indices: Optional[List[int]] = None) -> Tuple[np.ndarray, np.ndarray]:
+        """
+        重置指定环境
+        
+        Args:
+            indices: 要重置的环境索引，None表示全部重置
+            
+        Returns:
+            states: (num_envs, 4, 4) 状态数组
+            score_features: (num_envs, 2) 分数特征数组
+        """
+        if indices is None:
+            indices = range(self.num_envs)
+        
+        for i in indices:
+            self.states[i] = self.envs[i].reset()
+            self.scores[i] = self.envs[i].accumulated_score
+            self.situational_scores[i] = self.envs[i].situational_score
+        
+        return self._get_batch_state()
+    
+    def reset_single(self, idx: int) -> Tuple[np.ndarray, np.ndarray]:
+        """重置单个环境"""
+        self.states[idx] = self.envs[idx].reset()
+        self.scores[idx] = self.envs[idx].accumulated_score
+        self.situational_scores[idx] = self.envs[idx].situational_score
+        return self.states[idx], self._get_score_features(idx)
+    
+    def step(self, actions: List[int]) -> List[Transition]:
+        """
+        并行执行动作
+        
+        Args:
+            actions: 每个环境要执行的动作列表
+            
+        Returns:
+            transitions: 状态转移列表
+        """
+        transitions = []
+        
+        for i, action in enumerate(actions):
+            old_state = self.states[i].copy()
+            old_scores = self._get_score_features(i)
+            old_situational = self.situational_scores[i]
+            
+            # 执行动作
+            new_state, reward, moved, done = self.envs[i].move(action)
+            
+            # 更新状态
+            self.states[i] = new_state
+            self.scores[i] = self.envs[i].accumulated_score
+            self.situational_scores[i] = self.envs[i].situational_score
+            
+            # 创建转移记录
+            transition = Transition(
+                state=old_state,
+                scores=old_scores,
+                action=action,
+                reward=reward,
+                next_state=new_state.copy(),
+                next_scores=self._get_score_features(i),
+                done=done,
+                log_prob=0.0,  # 需要在外部填充
+                value=0.0,     # 需要在外部填充
+                valid_actions=self.envs[i].get_valid_actions()
+            )
+            transitions.append(transition)
+            
+            # 如果游戏结束，记录统计并重置
+            if done:
+                # 记录游戏统计
+                game_stats = {
+                    'score': self.scores[i],
+                    'situational_score': self.situational_scores[i],
+                    'max_tile': self.envs[i].get_max_tile(),
+                    'moves': self.envs[i].moves_count
+                }
+                self.games_completed += 1
+                
+                # 调用回调
+                if self.on_game_end:
+                    self.on_game_end(game_stats)
+                
+                self.reset_single(i)
+        
+        return transitions
+    
+    def _get_batch_state(self) -> Tuple[np.ndarray, np.ndarray]:
+        """获取批量状态"""
+        states = np.array(self.states, dtype=np.float32)
+        score_features = np.array([
+            self._get_score_features(i) for i in range(self.num_envs)
+        ], dtype=np.float32)
+        return states, score_features
+    
+    def _get_score_features(self, idx: int) -> np.ndarray:
+        """获取单个环境的分数特征"""
+        max_accumulated = 50000
+        max_situational = 200
+        return np.array([
+            min(self.scores[idx] / max_accumulated, 1.0),
+            min(self.situational_scores[idx] / max_situational, 1.0)
+        ], dtype=np.float32)
+    
+    def get_valid_actions(self) -> np.ndarray:
+        """获取所有环境的有效动作"""
+        return np.array([env.get_valid_actions() for env in self.envs])
+    
+    def get_game_stats(self) -> List[dict]:
+        """获取所有游戏的统计信息"""
+        return [
+            {
+                'score': env.accumulated_score,
+                'situational_score': env.situational_score,
+                'max_tile': env.get_max_tile(),
+                'moves': env.moves_count,
+                'game_over': env.game_over
+            }
+            for env in self.envs
+        ]
+    
+    def close(self):
+        """关闭环境"""
+        self.executor.shutdown(wait=False)
+
+
+class TrainingWorker:
+    """
+    训练工作器
+    负责收集轨迹数据
+    """
+    
+    def __init__(
+        self,
+        model,
+        env: ParallelGameEnv,
+        device: str = "cpu"
+    ):
+        self.model = model.to(device)
+        self.model.eval()
+        self.env = env
+        self.device = device
+        
+        self.stats = TrainingStats()
+    
+    def collect_trajectories(
+        self,
+        num_steps: int = 256,
+        deterministic: bool = False
+    ) -> List[Transition]:
+        """
+        收集轨迹数据
+        
+        Args:
+            num_steps: 每个环境收集的步数
+            deterministic: 是否确定性选择动作
+            
+        Returns:
+            transitions: 收集的转移数据
+        """
+        all_transitions = []
+        
+        for _ in range(num_steps):
+            # 获取当前状态
+            states = np.array(self.env.states, dtype=np.float32)
+            score_features = np.array([
+                self.env._get_score_features(i) 
+                for i in range(self.env.num_envs)
+            ], dtype=np.float32)
+            valid_actions = self.env.get_valid_actions()
+            
+            # 转换为张量
+            states_t = torch.FloatTensor(states).to(self.device)
+            scores_t = torch.FloatTensor(score_features).to(self.device)
+            valid_t = torch.BoolTensor(valid_actions).to(self.device)
+            
+            # 选择动作
+            actions = []
+            log_probs = []
+            values = []
+            
+            with torch.no_grad():
+                for i in range(self.env.num_envs):
+                    action, log_prob, value = self.model.get_action(
+                        states_t[i:i+1],
+                        scores_t[i:i+1],
+                        valid_t[i:i+1],
+                        deterministic=deterministic
+                    )
+                    actions.append(action)
+                    log_probs.append(log_prob)
+                    values.append(value)
+            
+            # 执行动作
+            transitions = self.env.step(actions)
+            
+            # 填充log_prob和value
+            for i, t in enumerate(transitions):
+                t.log_prob = log_probs[i]
+                t.value = values[i]
+                all_transitions.append(t)
+        
+        return all_transitions
+    
+    def run_episode(
+        self,
+        deterministic: bool = True,
+        max_steps: int = 10000
+    ) -> dict:
+        """
+        运行一局演示游戏
+        
+        Args:
+            deterministic: 是否确定性选择
+            max_steps: 最大步数
+            
+        Returns:
+            游戏统计信息
+        """
+        # 重置环境
+        env = Game2048()
+        state = env.reset()
+        
+        total_reward = 0
+        steps = 0
+        
+        while not env.game_over and steps < max_steps:
+            # 获取分数特征
+            max_accumulated = 50000
+            max_situational = 200
+            scores = np.array([
+                min(env.accumulated_score / max_accumulated, 1.0),
+                min(env.situational_score / max_situational, 1.0)
+            ], dtype=np.float32)
+            
+            # 获取有效动作
+            valid_actions = env.get_valid_actions()
+            
+            # 转换为张量
+            state_t = torch.FloatTensor(state).unsqueeze(0).to(self.device)
+            scores_t = torch.FloatTensor(scores).unsqueeze(0).to(self.device)
+            valid_t = torch.BoolTensor(valid_actions).unsqueeze(0).to(self.device)
+            
+            # 选择动作
+            with torch.no_grad():
+                action, _, _ = self.model.get_action(
+                    state_t, scores_t, valid_t, deterministic=deterministic
+                )
+            
+            # 执行动作
+            state, reward, moved, done = env.move(action)
+            total_reward += reward
+            steps += 1
+        
+        return {
+            'score': env.accumulated_score,
+            'situational_score': env.situational_score,
+            'max_tile': env.get_max_tile(),
+            'steps': steps,
+            'total_reward': total_reward
+        }
+
+
+class TrainingLoop:
+    """
+    完整的训练循环
+    """
+    
+    def __init__(
+        self,
+        model,
+        trainer,
+        num_envs: int = 4,
+        device: str = "cpu",
+        steps_per_update: int = 256,
+        save_interval: int = 100,
+        checkpoint_dir: str = "checkpoints"
+    ):
+        self.model = model
+        self.trainer = trainer
+        self.num_envs = num_envs
+        self.device = device
+        self.steps_per_update = steps_per_update
+        self.save_interval = save_interval
+        self.checkpoint_dir = checkpoint_dir
+        
+        # 初始化环境和工作器
+        self.env = ParallelGameEnv(num_envs=num_envs)
+        self.worker = TrainingWorker(model, self.env, device)
+        
+        # 训练状态
+        self.training = False
+        self.paused = False
+        self.stats = TrainingStats()
+        
+        # 回调函数
+        self.on_update_callback = None
+        self.on_game_end_callback = None
+    
+    def train(
+        self,
+        total_games: int = 10000,
+        stop_threshold: int = 100,
+        min_improvement: float = 0.01
+    ) -> None:
+        """
+        训练循环
+        
+        Args:
+            total_games: 总游戏局数
+            stop_threshold: 无提升停止阈值（局数）
+            min_improvement: 最小提升比例
+        """
+        self.training = True
+        
+        games_since_improvement = 0
+        best_avg_score = 0
+        
+        # 设置游戏结束回调
+        def on_game_end(game_stats):
+            self.stats.record_game(
+                score=game_stats['score'],
+                situational_score=game_stats['situational_score'],
+                max_tile=game_stats['max_tile'],
+                steps=game_stats['moves']
+            )
+            if self.on_game_end_callback:
+                self.on_game_end_callback(game_stats)
+        
+        self.env.on_game_end = on_game_end
+        
+        try:
+            while self.training and self.env.games_completed < total_games:
+                if self.paused:
+                    time.sleep(0.1)
+                    continue
+                
+                # 收集轨迹
+                transitions = self.worker.collect_trajectories(
+                    num_steps=self.steps_per_update // self.num_envs,
+                    deterministic=False
+                )
+                
+                # 更新模型
+                from trainer import RolloutBuffer
+                buffer = RolloutBuffer(capacity=len(transitions))
+                buffer.push_batch(transitions)
+                
+                update_stats = self.trainer.update(buffer)
+                
+                if self.on_update_callback:
+                    self.on_update_callback(update_stats)
+                
+                # 检查停止条件（每10次更新检查一次）
+                if self.env.games_completed % 10 == 0 and self.env.games_completed > 0:
+                    current_avg = self.stats.get_avg_stats(window=100)['avg_score']
+                    if current_avg > best_avg_score * (1 + min_improvement):
+                        best_avg_score = current_avg
+                        games_since_improvement = 0
+                    else:
+                        games_since_improvement = self.env.games_completed - int(best_avg_score / 100 * 100) if best_avg_score > 0 else 0
+                    
+                    if games_since_improvement >= stop_threshold:
+                        print(f"No improvement for {stop_threshold} updates, stopping.")
+                        break
+                
+        except KeyboardInterrupt:
+            print("Training interrupted by user.")
+        finally:
+            self.training = False
+            self.env.close()
+    
+    def stop(self) -> None:
+        """停止训练"""
+        self.training = False
+    
+    def pause(self) -> None:
+        """暂停训练"""
+        self.paused = True
+    
+    def resume(self) -> None:
+        """恢复训练"""
+        self.paused = False
+    
+    def get_stats(self) -> dict:
+        """获取当前统计信息"""
+        return self.stats.get_avg_stats()
+    
+    def save_checkpoint(self, path: str) -> None:
+        """保存模型检查点"""
+        torch.save({
+            'model_state_dict': self.model.state_dict(),
+            'optimizer_state_dict': self.trainer.optimizer.state_dict(),
+            'stats': self.stats.get_avg_stats(),
+            'games_played': self.stats.games_played
+        }, path)
+    
+    def load_checkpoint(self, path: str) -> None:
+        """加载模型检查点"""
+        checkpoint = torch.load(path, map_location=self.device)
+        self.model.load_state_dict(checkpoint['model_state_dict'])
+        self.trainer.optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
+
+
+if __name__ == "__main__":
+    from model import Game2048Transformer
+    from trainer import PPOTrainer
+    
+    # 测试并行环境
+    env = ParallelGameEnv(num_envs=4)
+    env.reset()
+    
+    print("Testing parallel environment...")
+    for i in range(10):
+        actions = [np.random.randint(0, 4) for _ in range(env.num_envs)]
+        transitions = env.step(actions)
+        print(f"Step {i}: collected {len(transitions)} transitions")
+    
+    env.close()
+    print("Parallel environment test passed!")

requirements.txt

@@ -0,0 +1,13 @@
+# 2048 AI Trainer 依赖
+
+# 深度学习框架
+torch>=2.0.0
+
+# 数值计算 (需要 <2 以兼容 torch)
+numpy<2
+
+# GUI
+PyQt5>=5.15.0
+
+# 绘图
+matplotlib>=3.7.0

train_simple.py

@@ -0,0 +1,149 @@
+"""
+简单训练脚本 - 直接训练并保存模型
+"""
+import os
+import sys
+import time
+import torch
+import numpy as np
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from game import Game2048
+from model import Game2048Transformer
+from trainer import PPOTrainer, RolloutBuffer
+
+def train_simple(
+    num_games: int = 1000,
+    save_path: str = "checkpoints/model.pt",
+    print_interval: int = 10
+):
+    """简单训练"""
+    print("=" * 50)
+    print("2048 AI Simple Training")
+    print("=" * 50)
+    
+    device = "cpu"
+    model = Game2048Transformer().to(device)
+    trainer = PPOTrainer(model, lr=3e-4, device=device)
+    
+    print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")
+    print(f"Training for {num_games} games...")
+    print("-" * 50)
+    
+    # 统计
+    scores = []
+    max_tiles = []
+    best_score = 0
+    
+    start_time = time.time()
+    
+    for game_idx in range(num_games):
+        # 运行一局游戏
+        game = Game2048()
+        game.reset()
+        
+        buffer = RolloutBuffer(capacity=10000)
+        
+        while not game.game_over:
+            state = game.get_state()
+            scores_feat = np.array([
+                min(game.accumulated_score / 50000, 1.0),
+                min(game.situational_score / 200, 1.0)
+            ], dtype=np.float32)
+            valid = game.get_valid_actions()
+            
+            # 转换张量
+            state_t = torch.FloatTensor(state).unsqueeze(0).to(device)
+            scores_t = torch.FloatTensor(scores_feat).unsqueeze(0).to(device)
+            valid_t = torch.BoolTensor(valid).unsqueeze(0).to(device)
+            
+            # 选择动作
+            with torch.no_grad():
+                action, log_prob, value = model.get_action(state_t, scores_t, valid_t, deterministic=False)
+            
+            # 执行动作
+            old_state = state.copy()
+            old_scores = scores_feat.copy()
+            
+            next_state, reward, moved, done = game.move(action)
+            
+            # 存储转移
+            from trainer import Transition
+            transition = Transition(
+                state=old_state,
+                scores=old_scores,
+                action=action,
+                reward=reward,
+                next_state=next_state.copy(),
+                next_scores=np.array([
+                    min(game.accumulated_score / 50000, 1.0),
+                    min(game.situational_score / 200, 1.0)
+                ], dtype=np.float32),
+                done=done,
+                log_prob=log_prob,
+                value=value,
+                valid_actions=valid
+            )
+            buffer.push(transition)
+            
+            # 每步更新
+            if len(buffer) >= 64:
+                trainer.update(buffer)
+                buffer.clear()
+        
+        # 记录结果
+        scores.append(game.accumulated_score)
+        max_tiles.append(game.get_max_tile())
+        
+        if game.accumulated_score > best_score:
+            best_score = game.accumulated_score
+        
+        # 打印进度
+        if (game_idx + 1) % print_interval == 0:
+            elapsed = time.time() - start_time
+            avg_score = np.mean(scores[-print_interval:])
+            avg_max_tile = np.mean(max_tiles[-print_interval:])
+            speed = (game_idx + 1) / elapsed
+            
+            print(
+                f"Game {game_idx + 1}/{num_games} | "
+                f"Avg Score: {avg_score:.0f} | "
+                f"Best: {best_score} | "
+                f"Max Tile: {avg_max_tile:.0f} | "
+                f"Speed: {speed:.2f} games/s"
+            )
+            
+            # 保存模型
+            os.makedirs(os.path.dirname(save_path), exist_ok=True)
+            torch.save({
+                'model_state_dict': model.state_dict(),
+                'game_idx': game_idx,
+                'best_score': best_score,
+                'avg_score': avg_score
+            }, save_path)
+    
+    # 最终保存
+    os.makedirs(os.path.dirname(save_path), exist_ok=True)
+    torch.save({
+        'model_state_dict': model.state_dict(),
+        'game_idx': num_games,
+        'best_score': best_score,
+        'avg_score': np.mean(scores[-100:])
+    }, save_path)
+    
+    elapsed = time.time() - start_time
+    print("\n" + "=" * 50)
+    print("Training Complete!")
+    print("=" * 50)
+    print(f"Total games: {num_games}")
+    print(f"Total time: {elapsed:.1f}s")
+    print(f"Average score (last 100): {np.mean(scores[-100:]):.0f}")
+    print(f"Best score: {best_score}")
+    print(f"Best max tile: {max(max_tiles)}")
+    print(f"Model saved to: {save_path}")
+    
+    return model
+
+if __name__ == "__main__":
+    train_simple(num_games=500)

trainer.py

@@ -0,0 +1,370 @@
+"""
+PPO训练器
+"""
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import torch.optim as optim
+import numpy as np
+from typing import List, Tuple, Optional
+from dataclasses import dataclass
+from collections import deque
+import random
+
+
+@dataclass
+class Transition:
+    """状态转移数据"""
+    state: np.ndarray      # 棋盘状态 (4, 4)
+    scores: np.ndarray     # 分数特征 (2,)
+    action: int            # 采取的动作
+    reward: float          # 奖励
+    next_state: np.ndarray # 下一状态
+    next_scores: np.ndarray # 下一分数
+    done: bool             # 是否结束
+    log_prob: float        # 动作的log概率
+    value: float           # 状态价值
+    valid_actions: np.ndarray  # 有效动作mask
+
+
+class RolloutBuffer:
+    """存储轨迹数据的缓冲区"""
+    
+    def __init__(self, capacity: int = 10000):
+        self.capacity = capacity
+        self.buffer: List[Transition] = []
+        self.position = 0
+    
+    def push(self, transition: Transition) -> None:
+        """添加一个转移"""
+        if len(self.buffer) < self.capacity:
+            self.buffer.append(transition)
+        else:
+            self.buffer[self.position] = transition
+        self.position = (self.position + 1) % self.capacity
+    
+    def push_batch(self, transitions: List[Transition]) -> None:
+        """批量添加转移"""
+        for t in transitions:
+            self.push(t)
+    
+    def get_all(self) -> List[Transition]:
+        """获取所有数据"""
+        return self.buffer.copy()
+    
+    def clear(self) -> None:
+        """清空缓冲区"""
+        self.buffer = []
+        self.position = 0
+    
+    def __len__(self) -> int:
+        return len(self.buffer)
+
+
+class PPOTrainer:
+    """PPO训练器"""
+    
+    def __init__(
+        self,
+        model,
+        lr: float = 1e-4,
+        gamma: float = 0.99,
+        gae_lambda: float = 0.95,
+        clip_ratio: float = 0.2,
+        value_coef: float = 0.5,
+        entropy_coef: float = 0.01,
+        max_grad_norm: float = 0.5,
+        update_epochs: int = 4,
+        batch_size: int = 64,
+        device: str = "cpu"
+    ):
+        self.model = model.to(device)
+        self.device = device
+        
+        self.gamma = gamma
+        self.gae_lambda = gae_lambda
+        self.clip_ratio = clip_ratio
+        self.value_coef = value_coef
+        self.entropy_coef = entropy_coef
+        self.max_grad_norm = max_grad_norm
+        self.update_epochs = update_epochs
+        self.batch_size = batch_size
+        
+        self.optimizer = optim.Adam(model.parameters(), lr=lr)
+        
+        # 训练统计
+        self.stats = {
+            'policy_loss': deque(maxlen=100),
+            'value_loss': deque(maxlen=100),
+            'entropy': deque(maxlen=100),
+            'total_loss': deque(maxlen=100)
+        }
+    
+    def compute_gae(
+        self,
+        rewards: np.ndarray,
+        values: np.ndarray,
+        dones: np.ndarray,
+        next_value: float = 0.0
+    ) -> Tuple[np.ndarray, np.ndarray]:
+        """
+        计算Generalized Advantage Estimation (GAE)
+        
+        Args:
+            rewards: 奖励序列 (T,)
+            values: 价值序列 (T,)
+            dones: 结束标志序列 (T,)
+            next_value: 最后状态的下一个价值
+            
+        Returns:
+            returns: 回报 (T,)
+            advantages: 优势 (T,)
+        """
+        T = len(rewards)
+        advantages = np.zeros(T, dtype=np.float32)
+        returns = np.zeros(T, dtype=np.float32)
+        
+        last_gae = 0
+        last_return = next_value
+        
+        for t in reversed(range(T)):
+            if dones[t]:
+                next_value_t = 0
+                last_gae = 0
+            else:
+                next_value_t = values[t + 1] if t + 1 < T else next_value
+            
+            delta = rewards[t] + self.gamma * next_value_t - values[t]
+            last_gae = delta + self.gamma * self.gae_lambda * (1 - dones[t]) * last_gae
+            advantages[t] = last_gae
+            
+            last_return = rewards[t] + self.gamma * (1 - dones[t]) * last_return
+            returns[t] = last_return
+        
+        return returns, advantages
+    
+    def update(self, buffer: RolloutBuffer) -> dict:
+        """
+        使用PPO更新模型
+        
+        Args:
+            buffer: 存储轨迹数据的缓冲区
+            
+        Returns:
+            训练统计信息
+        """
+        if len(buffer) < self.batch_size:
+            return {}
+        
+        # 获取所有数据
+        transitions = buffer.get_all()
+        
+        # 转换为数组
+        states = np.array([t.state for t in transitions])
+        scores = np.array([t.scores for t in transitions])
+        actions = np.array([t.action for t in transitions])
+        rewards = np.array([t.reward for t in transitions])
+        dones = np.array([t.done for t in transitions])
+        old_log_probs = np.array([t.log_prob for t in transitions])
+        old_values = np.array([t.value for t in transitions])
+        valid_actions = np.array([t.valid_actions for t in transitions])
+        
+        # 计算优势和回报
+        returns, advantages = self.compute_gae(rewards, old_values, dones)
+        
+        # 标准化优势
+        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
+        
+        # 转换为张量
+        states_t = torch.FloatTensor(states).to(self.device)
+        scores_t = torch.FloatTensor(scores).to(self.device)
+        actions_t = torch.LongTensor(actions).to(self.device)
+        old_log_probs_t = torch.FloatTensor(old_log_probs).to(self.device)
+        returns_t = torch.FloatTensor(returns).to(self.device)
+        advantages_t = torch.FloatTensor(advantages).to(self.device)
+        valid_actions_t = torch.BoolTensor(valid_actions).to(self.device)
+        
+        # PPO更新
+        total_policy_loss = 0
+        total_value_loss = 0
+        total_entropy = 0
+        num_updates = 0
+        
+        dataset_size = len(transitions)
+        indices = np.arange(dataset_size)
+        
+        for _ in range(self.update_epochs):
+            np.random.shuffle(indices)
+            
+            for start in range(0, dataset_size, self.batch_size):
+                end = start + self.batch_size
+                batch_indices = indices[start:end]
+                
+                # 获取批次数据
+                batch_states = states_t[batch_indices]
+                batch_scores = scores_t[batch_indices]
+                batch_actions = actions_t[batch_indices]
+                batch_old_log_probs = old_log_probs_t[batch_indices]
+                batch_returns = returns_t[batch_indices]
+                batch_advantages = advantages_t[batch_indices]
+                batch_valid = valid_actions_t[batch_indices]
+                
+                # 前向传播
+                log_probs, values, entropy = self.model.evaluate_actions(
+                    batch_states, batch_actions, batch_scores, batch_valid
+                )
+                
+                # 策略损失 (PPO Clip)
+                ratio = torch.exp(log_probs - batch_old_log_probs)
+                surr1 = ratio * batch_advantages
+                surr2 = torch.clamp(ratio, 1 - self.clip_ratio, 1 + self.clip_ratio) * batch_advantages
+                policy_loss = -torch.min(surr1, surr2).mean()
+                
+                # 价值损失
+                value_loss = F.mse_loss(values.squeeze(), batch_returns)
+                
+                # 总损失
+                loss = (
+                    policy_loss +
+                    self.value_coef * value_loss -
+                    self.entropy_coef * entropy.mean()
+                )
+                
+                # 反向传播
+                self.optimizer.zero_grad()
+                loss.backward()
+                nn.utils.clip_grad_norm_(self.model.parameters(), self.max_grad_norm)
+                self.optimizer.step()
+                
+                total_policy_loss += policy_loss.item()
+                total_value_loss += value_loss.item()
+                total_entropy += entropy.mean().item()
+                num_updates += 1
+        
+        # 记录统计
+        stats = {
+            'policy_loss': total_policy_loss / num_updates,
+            'value_loss': total_value_loss / num_updates,
+            'entropy': total_entropy / num_updates
+        }
+        
+        for key, value in stats.items():
+            self.stats[key].append(value)
+        
+        return stats
+    
+    def get_recent_stats(self) -> dict:
+        """获取最近的训练统计"""
+        return {key: np.mean(values) for key, values in self.stats.items() if values}
+
+
+class TrainingStats:
+    """训练统计记录器"""
+    
+    def __init__(self):
+        self.games_played = 0
+        self.total_steps = 0
+        self.scores = []           # 每局累积分数
+        self.situational_scores = []  # 每局平均局面分数
+        self.max_tiles = []        # 每局最大砖块
+        self.game_lengths = []     # 每局步数
+        
+        # 历史记录用于绘图
+        self.score_history = []
+        self.situational_history = []
+        self.max_tile_history = []
+        self.steps_history = []
+        
+        # 最佳记录
+        self.best_score = 0
+        self.best_max_tile = 0
+    
+    def record_game(
+        self, 
+        score: int, 
+        situational_score: float,
+        max_tile: int, 
+        steps: int
+    ) -> None:
+        """记录一局游戏"""
+        self.games_played += 1
+        self.total_steps += steps
+        
+        self.scores.append(score)
+        self.situational_scores.append(situational_score)
+        self.max_tiles.append(max_tile)
+        self.game_lengths.append(steps)
+        
+        self.score_history.append(score)
+        self.situational_history.append(situational_score)
+        self.max_tile_history.append(max_tile)
+        self.steps_history.append(steps)
+        
+        if score > self.best_score:
+            self.best_score = score
+        if max_tile > self.best_max_tile:
+            self.best_max_tile = max_tile
+    
+    def get_avg_stats(self, window: int = 100) -> dict:
+        """获取平均统计"""
+        def avg(lst):
+            if not lst:
+                return 0
+            recent = lst[-window:]
+            return sum(recent) / len(recent)
+        
+        return {
+            'games_played': self.games_played,
+            'total_steps': self.total_steps,
+            'avg_score': avg(self.scores),
+            'avg_situational': avg(self.situational_scores),
+            'avg_max_tile': avg(self.max_tiles),
+            'avg_game_length': avg(self.game_lengths),
+            'best_score': self.best_score,
+            'best_max_tile': self.best_max_tile,
+            'recent_scores': self.scores[-10:] if self.scores else [],
+            'recent_max_tiles': self.max_tiles[-10:] if self.max_tiles else []
+        }
+
+
+if __name__ == "__main__":
+    from model import Game2048Transformer
+    
+    # 测试PPO训练器
+    device = torch.device("cpu")
+    model = Game2048Transformer().to(device)
+    trainer = PPOTrainer(model, device=device)
+    
+    # 创建测试数据
+    buffer = RolloutBuffer(capacity=1000)
+    
+    for _ in range(100):
+        t = Transition(
+            state=np.random.randn(4, 4).astype(np.float32),
+            scores=np.random.rand(2).astype(np.float32),
+            action=np.random.randint(0, 4),
+            reward=np.random.randn(),
+            next_state=np.random.randn(4, 4).astype(np.float32),
+            next_scores=np.random.rand(2).astype(np.float32),
+            done=np.random.rand() < 0.1,
+            log_prob=np.random.randn(),
+            value=np.random.randn(),
+            valid_actions=np.ones(4, dtype=bool)
+        )
+        buffer.push(t)
+    
+    # 测试更新
+    stats = trainer.update(buffer)
+    print(f"Training stats: {stats}")
+    
+    # 测试统计
+    training_stats = TrainingStats()
+    for i in range(10):
+        training_stats.record_game(
+            score=1000 * (i + 1),
+            situational_score=50.0 + i * 5,
+            max_tile=2 ** (i + 5),
+            steps=100 + i * 10
+        )
+    
+    print(f"Average stats: {training_stats.get_avg_stats()}")

utils.py

@@ -0,0 +1,295 @@
+"""
+工具函数
+"""
+import os
+import json
+import numpy as np
+import torch
+from datetime import datetime
+from typing import Dict, Any, Optional
+import shutil
+
+
+def ensure_dir(path: str) -> str:
+    """确保目录存在，不存在则创建"""
+    if not os.path.exists(path):
+        os.makedirs(path)
+    return path
+
+
+def save_checkpoint(
+    model: torch.nn.Module,
+    optimizer: torch.optim.Optimizer,
+    epoch: int,
+    stats: Dict[str, Any],
+    path: str
+) -> None:
+    """保存训练检查点"""
+    ensure_dir(os.path.dirname(path))
+    torch.save({
+        'epoch': epoch,
+        'model_state_dict': model.state_dict(),
+        'optimizer_state_dict': optimizer.state_dict(),
+        'stats': stats,
+        'timestamp': datetime.now().isoformat()
+    }, path)
+
+
+def load_checkpoint(
+    path: str,
+    model: torch.nn.Module,
+    optimizer: Optional[torch.optim.Optimizer] = None,
+    device: str = 'cpu'
+) -> Dict[str, Any]:
+    """加载训练检查点"""
+    checkpoint = torch.load(path, map_location=device, weights_only=False)
+    model.load_state_dict(checkpoint['model_state_dict'])
+    
+    if optimizer is not None and 'optimizer_state_dict' in checkpoint:
+        optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
+    
+    return checkpoint
+
+
+def save_training_log(log_data: Dict[str, Any], path: str) -> None:
+    """保存训练日志"""
+    ensure_dir(os.path.dirname(path))
+    
+    # 读取现有日志
+    if os.path.exists(path):
+        with open(path, 'r', encoding='utf-8') as f:
+            try:
+                logs = json.load(f)
+            except json.JSONDecodeError:
+                logs = []
+    else:
+        logs = []
+    
+    # 添加新记录
+    log_data['timestamp'] = datetime.now().isoformat()
+    logs.append(log_data)
+    
+    # 保存
+    with open(path, 'w', encoding='utf-8') as f:
+        json.dump(logs, f, indent=2, ensure_ascii=False)
+
+
+def format_time(seconds: float) -> str:
+    """格式化时间"""
+    if seconds < 60:
+        return f'{seconds:.1f}s'
+    elif seconds < 3600:
+        minutes = seconds / 60
+        return f'{minutes:.1f}m'
+    else:
+        hours = seconds / 3600
+        return f'{hours:.1f}h'
+
+
+def format_number(num: int) -> str:
+    """格式化数字（添加逗号分隔）"""
+    return f'{num:,}'
+
+
+def calculate_ema(values: list, alpha: float = 0.1) -> list:
+    """计算指数移动平均"""
+    if not values:
+        return []
+    
+    ema = [values[0]]
+    for value in values[1:]:
+        ema.append(alpha * value + (1 - alpha) * ema[-1])
+    return ema
+
+
+def get_tile_color(value: int) -> str:
+    """获取砖块颜色"""
+    colors = {
+        0: '#cdc1b4',
+        2: '#eee4da',
+        4: '#ede0c8',
+        8: '#f2b179',
+        16: '#f59563',
+        32: '#f67c5f',
+        64: '#f65e3b',
+        128: '#edcf72',
+        256: '#edcc61',
+        512: '#edc850',
+        1024: '#edc53f',
+        2048: '#edc22e',
+    }
+    return colors.get(value, '#3c3a32')
+
+
+def get_text_color(value: int) -> str:
+    """获取文字颜色"""
+    if value <= 4:
+        return '#776e65'
+    return '#f9f6f2'
+
+
+class EarlyStopping:
+    """早停机制"""
+    
+    def __init__(
+        self,
+        patience: int = 100,
+        min_delta: float = 0.01,
+        mode: str = 'max'
+    ):
+        """
+        Args:
+            patience: 容忍的epoch数
+            min_delta: 最小改进
+            mode: 'max' 或 'min'
+        """
+        self.patience = patience
+        self.min_delta = min_delta
+        self.mode = mode
+        self.counter = 0
+        self.best_value = None
+        self.should_stop = False
+    
+    def __call__(self, value: float) -> bool:
+        """
+        检查是否应该停止
+        
+        Args:
+            value: 当前值
+            
+        Returns:
+            是否应该停止
+        """
+        if self.best_value is None:
+            self.best_value = value
+            return False
+        
+        if self.mode == 'max':
+            improved = value > self.best_value + self.min_delta
+        else:
+            improved = value < self.best_value - self.min_delta
+        
+        if improved:
+            self.best_value = value
+            self.counter = 0
+        else:
+            self.counter += 1
+            if self.counter >= self.patience:
+                self.should_stop = True
+        
+        return self.should_stop
+
+
+class MetricTracker:
+    """指标跟踪器"""
+    
+    def __init__(self, window_size: int = 100):
+        self.window_size = window_size
+        self.metrics = {}
+    
+    def update(self, name: str, value: float) -> None:
+        """更新指标"""
+        if name not in self.metrics:
+            self.metrics[name] = []
+        self.metrics[name].append(value)
+        
+        # 保持窗口大小
+        if len(self.metrics[name]) > self.window_size:
+            self.metrics[name] = self.metrics[name][-self.window_size:]
+    
+    def get_mean(self, name: str) -> float:
+        """获取平均值"""
+        if name not in self.metrics or not self.metrics[name]:
+            return 0.0
+        return np.mean(self.metrics[name])
+    
+    def get_std(self, name: str) -> float:
+        """获取标准差"""
+        if name not in self.metrics or len(self.metrics[name]) < 2:
+            return 0.0
+        return np.std(self.metrics[name])
+    
+    def get_all_means(self) -> Dict[str, float]:
+        """获取所有指标的平均值"""
+        return {name: self.get_mean(name) for name in self.metrics}
+
+
+def set_seed(seed: int) -> None:
+    """设置随机种子"""
+    import random
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed(seed)
+        torch.cuda.manual_seed_all(seed)
+
+
+def get_device() -> str:
+    """获取可用设备"""
+    if torch.cuda.is_available():
+        return 'cuda'
+    return 'cpu'
+
+
+def count_parameters(model: torch.nn.Module) -> int:
+    """计算模型参数数量"""
+    return sum(p.numel() for p in model.parameters() if p.requires_grad)
+
+
+def print_model_info(model: torch.nn.Module) -> None:
+    """打印模型信息"""
+    total_params = count_parameters(model)
+    print(f"模型参数数量: {format_number(total_params)}")
+    print(f"模型大小: {total_params * 4 / 1024 / 1024:.2f} MB (float32)")
+
+
+def export_to_onnx(
+    model: torch.nn.Module,
+    path: str,
+    input_size: tuple = (1, 4, 4)
+) -> None:
+    """导出模型到ONNX格式"""
+    model.eval()
+    dummy_input = torch.randn(*input_size)
+    dummy_scores = torch.randn(1, 2)
+    dummy_mask = torch.ones(1, 4, dtype=torch.bool)
+    
+    ensure_dir(os.path.dirname(path))
+    torch.onnx.export(
+        model,
+        (dummy_input, dummy_scores, dummy_mask),
+        path,
+        input_names=['board', 'scores', 'mask'],
+        output_names=['policy', 'value'],
+        dynamic_axes={
+            'board': {0: 'batch_size'},
+            'scores': {0: 'batch_size'},
+            'mask': {0: 'batch_size'}
+        }
+    )
+    print(f"模型已导出到: {path}")
+
+
+if __name__ == "__main__":
+    # 测试工具函数
+    print("Testing utility functions...")
+    
+    # 测试时间格式化
+    print(f"Format time: {format_time(45.5)}, {format_time(125.3)}, {format_time(3661)}")
+    
+    # 测试数字格式化
+    print(f"Format number: {format_number(1234567)}")
+    
+    # 测试EMA
+    values = [1, 2, 3, 4, 5]
+    print(f"EMA: {calculate_ema(values)}")
+    
+    # 测试早停
+    early_stop = EarlyStopping(patience=3, min_delta=0.1)
+    scores = [10, 11, 12, 12, 12, 12, 12]
+    for i, score in enumerate(scores):
+        stop = early_stop(score)
+        print(f"Epoch {i}: score={score}, stop={stop}")
+    
+    print("All tests passed!")