本文目录导读：
我来为你介绍几个Python用强化学习玩游戏的经典案例,从简单到复杂：
基础框架：CartPole游戏（OpenAI Gym）
安装依赖
pip install gym tensorflow keras-rl2 pip install pygame
案例1：使用DQN玩CartPole
import gym
import numpy as np
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Flatten
from tensorflow.keras.optimizers import Adam
from rl.agents import DQNAgent
from rl.policy import BoltzmannQPolicy
from rl.memory import SequentialMemory
# 创建环境
env = gym.make('CartPole-v1')
states = env.observation_space.shape[0]
actions = env.action_space.n
# 构建神经网络模型
def build_model(states, actions):
model = Sequential()
model.add(Flatten(input_shape=(1, states)))
model.add(Dense(24, activation='relu'))
model.add(Dense(24, activation='relu'))
model.add(Dense(actions, activation='linear'))
return model
model = build_model(states, actions)
# 构建DQN代理
def build_agent(model, actions):
policy = BoltzmannQPolicy()
memory = SequentialMemory(limit=50000, window_length=1)
dqn = DQNAgent(model=model, memory=memory, policy=policy,
nb_actions=actions, nb_steps_warmup=10,
target_model_update=1e-2)
return dqn
dqn = build_agent(model, actions)
dqn.compile(Adam(lr=1e-3), metrics=['mae'])
# 训练
dqn.fit(env, nb_steps=50000, visualize=False, verbose=1)
# 测试
scores = dqn.test(env, nb_episodes=10, visualize=True)
print(f'平均得分: {np.mean(scores.history["episode_reward"])}')
经典游戏：打砖块（Breakout）
使用深度强化学习
import gym
import tensorflow as tf
from tensorflow.keras import layers
import numpy as np
class BreakoutAgent:
def __init__(self, env):
self.env = env
self.state_size = (84, 84, 4) # 84x84的灰度图，4帧堆叠
self.action_size = env.action_space.n
self.model = self.build_model()
def build_model(self):
"""构建CNN模型"""
model = tf.keras.Sequential([
layers.Conv2D(32, (8, 8), strides=4, activation='relu',
input_shape=self.state_size),
layers.Conv2D(64, (4, 4), strides=2, activation='relu'),
layers.Conv2D(64, (3, 3), strides=1, activation='relu'),
layers.Flatten(),
layers.Dense(512, activation='relu'),
layers.Dense(self.action_size, activation='linear')
])
model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(0.00025))
return model
# 使用现成的强化学习库
from stable_baselines3 import DQN, PPO
from stable_baselines3.common.vec_env import DummyVecEnv
# 更简单的实现方式
def train_breakout():
# 创建环境
env = gym.make('Breakout-v4')
env = DummyVecEnv([lambda: env])
# 使用PPO算法
model = PPO('CnnPolicy', env, verbose=1,
learning_rate=0.0001,
n_steps=2048,
batch_size=64)
# 训练
model.learn(total_timesteps=100000)
# 保存模型
model.save("breakout_ppo")
# 测试
obs = env.reset()
for _ in range(1000):
action, _states = model.predict(obs, deterministic=True)
obs, rewards, dones, info = env.step(action)
env.render()
env.close()
# train_breakout()
自定义游戏：Flappy Bird
从零实现强化学习
import pygame
import random
import numpy as np
from collections import deque
import tensorflow as tf
from tensorflow.keras import layers
# 简化的Flappy Bird环境
class FlappyBirdEnv:
def __init__(self):
pygame.init()
self.width = 288
self.height = 512
self.screen = pygame.display.set_mode((self.width, self.height))
self.clock = pygame.time.Clock()
self.bird_y = self.height // 2
self.bird_velocity = 0
self.gravity = 0.5
self.jump_strength = -8
self.pipe_width = 52
self.pipe_gap = 100
self.pipe_x = self.width
self.pipe_height = random.randint(100, 400)
self.score = 0
self.game_over = False
def reset(self):
self.bird_y = self.height // 2
self.bird_velocity = 0
self.pipe_x = self.width
self.pipe_height = random.randint(100, 400)
self.score = 0
self.game_over = False
return self.get_state()
def get_state(self):
# 返回状态：相对位置、速度等
return np.array([
self.bird_y / self.height,
self.bird_velocity / 10,
(self.pipe_x - 50) / self.width,
(self.pipe_height - 100) / 400
])
def step(self, action):
# action: 0=不操作, 1=跳跃
if action == 1:
self.bird_velocity = self.jump_strength
# 物理更新
self.bird_velocity += self.gravity
self.bird_y += self.bird_velocity
self.pipe_x -= 3
# 碰撞检测
reward = 0.1
if (self.bird_y < 0 or self.bird_y > self.height or
(self.pipe_x < 50 < self.pipe_x + self.pipe_width and
(self.bird_y < self.pipe_height or
self.bird_y > self.pipe_height + self.pipe_gap))):
self.game_over = True
reward = -10
# 通过管道
if self.pipe_x + self.pipe_width < 0:
self.score += 1
reward = 10
self.pipe_x = self.width
self.pipe_height = random.randint(100, 400)
return self.get_state(), reward, self.game_over
# DQN代理
class DQNAgent:
def __init__(self, state_size, action_size):
self.state_size = state_size
self.action_size = action_size
self.memory = deque(maxlen=2000)
self.gamma = 0.95
self.epsilon = 1.0
self.epsilon_min = 0.01
self.epsilon_decay = 0.995
self.learning_rate = 0.001
self.model = self._build_model()
def _build_model(self):
model = tf.keras.Sequential([
layers.Dense(24, input_shape=(self.state_size,), activation='relu'),
layers.Dense(24, activation='relu'),
layers.Dense(self.action_size, activation='linear')
])
model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(lr=self.learning_rate))
return model
def remember(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def act(self, state):
if np.random.rand() <= self.epsilon:
return random.randrange(self.action_size)
act_values = self.model.predict(state)
return np.argmax(act_values[0])
def replay(self, batch_size):
if len(self.memory) < batch_size:
return
minibatch = random.sample(self.memory, batch_size)
for state, action, reward, next_state, done in minibatch:
target = reward
if not done:
target = reward + self.gamma * np.amax(self.model.predict(next_state)[0])
target_f = self.model.predict(state)
target_f[0][action] = target
self.model.fit(state, target_f, epochs=1, verbose=0)
if self.epsilon > self.epsilon_min:
self.epsilon *= self.epsilon_decay
# 训练函数
def train_flappy_bird():
env = FlappyBirdEnv()
state_size = 4
action_size = 2
agent = DQNAgent(state_size, action_size)
episodes = 1000
batch_size = 32
for e in range(episodes):
state = env.reset()
state = np.reshape(state, [1, state_size])
for time in range(500):
action = agent.act(state)
next_state, reward, done = env.step(action)
next_state = np.reshape(next_state, [1, state_size])
agent.remember(state, action, reward, next_state, done)
state = next_state
if done:
print(f"Episode {e+1}/{episodes}, Score: {env.score}")
break
if len(agent.memory) > batch_size:
agent.replay(batch_size)
# 保存模型
agent.model.save("flappy_bird_dqn.h5")
print("训练完成！")
# train_flappy_bird()
使用预训练模型玩游戏
# 更实用的方式：使用预训练模型
from stable_baselines3 import DQN
import gym
def play_with_pretrained():
# 加载预训练模型（需要先下载）
# 实际使用时可以从网上下载预训练权重
model = DQN.load("path_to_pretrained_model")
env = gym.make('CartPole-v1')
obs = env.reset()
while True:
action, _states = model.predict(obs, deterministic=True)
obs, rewards, dones, info = env.step(action)
env.render()
if dones:
obs = env.reset()
# play_with_pretrained()
进阶：使用Ray实现分布式强化学习
# 安装: pip install ray[rllib]
import ray
from ray import tune
from ray.rllib.agents.ppo import PPOTrainer
ray.init()
# 配置和训练
config = {
"env": "CartPole-v1",
"lr": 0.0001,
"num_workers": 4,
"train_batch_size": 4000,
}
# 自动调参训练
analysis = tune.run(
"PPO",
config=config,
stop={"episode_reward_mean": 200},
num_samples=3,
)
print("最佳配置:", analysis.get_best_config(metric="episode_reward_mean", mode="max"))
学习路径建议
-
入门顺序：
- 先理解CartPole（最简单的控制问题）
- 再尝试经典Atari游戏（如Breakout）
- 最后尝试自定义游戏
-
调试技巧：
- 使用TensorBoard监控训练过程
- 记录奖励曲线
- 可视化智能体的行为
-
资源推荐：
- OpenAI Gym: 标准游戏环境
- Stable-Baselines3: 成熟的算法库
- RLlib: 分布式强化学习
- CleanRL: 简单实现的强化学习算法
这些案例涵盖了从简单到复杂的强化学习应用,建议从CartPole开始，逐步加深理解，强化学习的训练可能需要大量时间和计算资源，可以先使用小规模环境进行测试。
