wsy182
/
mjAi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1 

1
dev
wsy182 2024-12-01 20:01:23 +08:00
parent c9defe78f1
commit 0864295a6e
4 changed files with 39 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
models/ppo_mahjong_model.zip
View File

Binary file not shown.

11
scripts/train_chengdu_mahjong_model.py
View File

@ -3,6 +3,7 @@ from stable_baselines3 import PPO
from src.environment.chengdu_mahjong_env import ChengduMahjongEnv
import torch
from configs.log_config import setup_logging
from loguru import logger # 添加 logger
def train_model():
# 创建 MahjongEnv 环境实例
@ -10,7 +11,7 @@ def train_model():
# 检查是否有可用的 GPU
device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"使用设备: {device}")
logger.info(f"使用设备: {device}") # 替换 print 为 logger.info
# 使用 PPO 算法训练模型，切换到 MultiInputPolicy
model = PPO(
@ -22,18 +23,22 @@ def train_model():
)
# 训练模型，训练总步数为 100000
model.learn(total_timesteps=100000)
logger.info("开始训练模型...")
model.learn(total_timesteps=100)
logger.info("模型训练完成！")
# 保存训练后的模型
model.save("../models/ppo_mahjong_model")
logger.info("模型已保存到 '../models/ppo_mahjong_model'")
# 测试模型
logger.info("开始测试模型...")
obs = env.reset()
done = False
while not done:
action, _states = model.predict(obs) # 使用训练好的模型来选择动作
obs, reward, done, info = env.step(action) # 执行动作
print(f"动作: {action}, 奖励: {reward}, 是否结束: {done}, 信息: {info}")
logger.info(f"动作: {action}, 奖励: {reward}, 是否结束: {done}, 信息: {info}") # 替换 print 为 logger.info
if __name__ == "__main__":
# 调用配置函数来设置日志

62
src/engine/chengdu_mahjong_engine.py
View File

@ -48,57 +48,27 @@ class ChengduMahjongEngine:
def play_turn(self):
current_player = self.state.current_player
if self.state.draw_counts[current_player] == 0:
# 判断庄家是否天胡
tianhu = self.state.can_win(
self.state.hands[current_player],
self.state.melds[current_player],
self.state.missing_suits[current_player],
)
# 玩家摸牌逻辑
draw_tile(self)
if tianhu:
# 天胡结算
self.state.winners.append(current_player)
self.state.print_game_state(current_player)
for i in range(4):
if i != current_player:
self.state.scores[i] -= self.state.bottom_score * 2 ** 5
self.state.scores[current_player] += self.state.bottom_score * 2 ** 5
else:
# 判断是否可以杠
if self.state.hands[current_player].can_gang():
# 获取杠牌类型
gang_type = self.state.hands[current_player].get_gang_type()
# 玩家选择一张牌打出
tile = random_choice(self.state.hands[current_player], self.state.missing_suits[current_player])
logger.info(f"玩家 {current_player} 选择打牌: {tile}")
# AI 决策是否杠牌
if should_gang(current_player, self.state, gang_type):
self.state.print_game_state(current_player)
# 检查其他玩家是否可以对该牌进行操作
actions_taken = self.check_other_players(tile)
# 计算杠牌得分
if gang_type == "暗杠":
# 暗杠分数结算
for i in range(4):
if i != current_player:
self.state.scores[i] -= self.state.bottom_score * 2 ** 2 # 扣分
self.state.scores[current_player] += self.state.bottom_score * 2 ** 2 * 3 # 加分
elif gang_type == "明杠":
# 明杠分数结算
for i in range(4):
if i != current_player:
self.state.scores[i] -= self.state.bottom_score * 2 ** 1 # 扣分
self.state.scores[current_player] += self.state.bottom_score * 2 ** 1 * 3 # 加分
if not actions_taken:
# 将牌加入弃牌堆
self.state.discards[current_player].append(tile)
logger.info(f"玩家 {current_player} 打出的牌 {tile} 没有触发其他玩家的操作")
logger.info(f"玩家 {current_player} 杠牌，类型: {gang_type}")
else:
# 当前玩家出牌
tile = random_choice(self.state.hands[current_player], self.state.missing_suits[current_player])
logger.info(f"玩家 {current_player} 打出牌: {tile}")
# 切换到下一位玩家
self.state.current_player = (current_player + 1) % 4
logger.info(f"轮到玩家 {self.state.current_player} 出牌")
# 检查其他玩家的反应
if not check_other_players(self,tile):
# 没有触发其他玩家操作，移动到下一个玩家
self.state.current_player = (current_player + 1) % 4
draw_tile(self)
# 检查游戏结束条件
self.check_game_over()
def check_game_over(self):
"""

26
src/environment/chengdu_mahjong_env.py
View File

@ -41,33 +41,35 @@ class ChengduMahjongEnv(gym.Env):
:return: obs, reward, done, info
"""
current_player = self.engine.state.current_player
hand = self.engine.state.hands[current_player].tiles # 当前玩家手牌
# **1. 检查动作是否合法并执行**
if action < 14: # 打牌动作
if action >= len(self.engine.state.hands[current_player].tiles):
raise ValueError(f"动作 {action} 超出手牌范围")
tile = self.engine.state.hands[current_player].tiles[action]
# **1. 执行动作并检查合法性**
if action < len(hand): # 打牌动作
tile = hand[action]
logger.info(f"玩家 {current_player} 选择打牌: {tile}")
self.engine.check_other_players(tile)
elif action == 14: # 碰
tile_to_peng = self._get_tile_for_special_action("peng")
if tile_to_peng:
handle_peng(self.engine,current_player, tile_to_peng)
handle_peng(self.engine, current_player, tile_to_peng)
logger.info(f"玩家 {current_player} 碰了牌: {tile_to_peng}")
else:
logger.warning("碰动作无效，未满足条件")
elif action == 15: # 杠
tile_to_gang = self._get_tile_for_special_action("gang")
if tile_to_gang:
handle_gang(current_player, tile_to_gang, mode="an") # 默认暗杠
handle_gang(self.engine, current_player, tile_to_gang, mode="an")
logger.info(f"玩家 {current_player} 杠了牌: {tile_to_gang}")
else:
logger.warning("杠动作无效，未满足条件")
elif action == 16: # 胡
if self.engine.state.can_win(
self.engine.state.hands[current_player],
self.engine.state.melds[current_player],
self.engine.state.missing_suits[current_player]
self.engine.state.hands[current_player],
self.engine.state.melds[current_player],
self.engine.state.missing_suits[current_player]
):
handle_win(current_player, None, None)
handle_win(self.engine, current_player, None, None)
logger.info(f"玩家 {current_player} 胡牌！")
else:
logger.warning("胡动作无效，未满足条件")
else:
@ -90,6 +92,8 @@ class ChengduMahjongEnv(gym.Env):
}
return obs, reward, done, info
def _get_observation(self):
"""
提取当前玩家的观察空间