Hands-on RL projects: game playing, robot control, and real-world applications.
Practical projects to learn RL by building functional systems.
Project 1: Game Playing with Q-Learning
Game: CartPole (balance pole on cart)
import gym
import numpy as np
class QLearningAgentCartPole:
def __init__(self, n_bins=20):
self.env = gym.make('CartPole-v1')
self.n_bins = n_bins
self.Q = np.zeros((n_bins**4, self.env.action_space.n))
# State discretization bounds
self.bounds = [
(-2.4, 2.4), # Cart position
(-3, 3), # Cart velocity
(-0.2, 0.2), # Pole angle
(-3, 3) # Pole angular velocity
]
def discretize_state(self, state):
"""Convert continuous state to discrete bins"""
indices = []
for i, (val, (low, high)) in enumerate(zip(state, self.bounds)):
bin_idx = int((val - low) / (high - low) * self.n_bins)
bin_idx = np.clip(bin_idx, 0, self.n_bins - 1)
indices.append(bin_idx)
# Convert to single index
return int(np.ravel_multi_index(indices, (self.n_bins,)*4))
def train(self, episodes=1000, alpha=0.1, gamma=0.99, epsilon=0.1):
"""Train Q-Learning agent"""
rewards_per_episode = []
for episode in range(episodes):
state = self.env.reset()
discrete_state = self.discretize_state(state)
done = False
total_reward = 0
while not done:
# Epsilon-greedy action selection
if np.random.random() < epsilon:
action = self.env.action_space.sample()
else:
action = np.argmax(self.Q[discrete_state, :])
# Take action
next_state, reward, done, _ = self.env.step(action)
next_discrete_state = self.discretize_state(next_state)
# Q-Learning update
max_next_Q = np.max(self.Q[next_discrete_state, :])
self.Q[discrete_state, action] += alpha * (
reward + gamma * max_next_Q - self.Q[discrete_state, action]
)
total_reward += reward
discrete_state = next_discrete_state
rewards_per_episode.append(total_reward)
if episode % 100 == 0:
avg_reward = np.mean(rewards_per_episode[-100:])
print(f"Episode {episode}: Avg Reward = {avg_reward:.1f}")
return rewards_per_episode
def test(self, episodes=10):
"""Test trained agent"""
total_rewards = []
for episode in range(episodes):
state = self.env.reset()
discrete_state = self.discretize_state(state)
done = False
total_reward = 0
while not done:
action = np.argmax(self.Q[discrete_state, :])
next_state, reward, done, _ = self.env.step(action)
next_discrete_state = self.discretize_state(next_state)
total_reward += reward
discrete_state = next_discrete_state
total_rewards.append(total_reward)
self.env.render()
print(f"Average test reward: {np.mean(total_rewards):.1f}")
return total_rewards
# Usage
agent = QLearningAgentCartPole()
rewards = agent.train(episodes=1000)
agent.test(episodes=5)
Project 2: Atari Game with DQN
import tensorflow as tf
import numpy as np
from collections import deque
class AtariDQN:
def __init__(self, env_name='Pong-v0'):
self.env = gym.make(env_name)
self.action_size = self.env.action_space.n
self.state_size = (84, 84, 4) # Preprocessed Atari frames
self.memory = deque(maxlen=100000)
self.epsilon = 1.0
self.epsilon_min = 0.01
self.epsilon_decay = 0.995
self.learning_rate = 0.00025
self.gamma = 0.99
self.model = self.build_model()
self.target_model = self.build_model()
self.update_target_model()
self.frame_count = 0
def build_model(self):
"""Build DQN network"""
model = tf.keras.Sequential([
tf.keras.layers.Conv2D(32, (8, 8), strides=(4, 4),
activation='relu', input_shape=self.state_size),
tf.keras.layers.Conv2D(64, (4, 4), strides=(2, 2),
activation='relu'),
tf.keras.layers.Conv2D(64, (3, 3), strides=(1, 1),
activation='relu'),
tf.keras.layers.Flatten(),
tf.keras.layers.Dense(512, activation='relu'),
tf.keras.layers.Dense(self.action_size)
])
model.compile(optimizer=tf.keras.optimizers.Adam(self.learning_rate),
loss='mse')
return model
def update_target_model(self):
self.target_model.set_weights(self.model.get_weights())
def preprocess_frame(self, frame):
"""Preprocess Atari frame"""
# Grayscale, resize to 84x84
frame = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
frame = cv2.resize(frame, (84, 84))
return frame
def remember(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def act(self, state):
self.frame_count += 1
if np.random.random() <= self.epsilon:
return self.env.action_space.sample()
q_values = self.model.predict(state, verbose=0)
return np.argmax(q_values[0])
def replay(self, batch_size):
if len(self.memory) < batch_size:
return
batch = np.random.choice(len(self.memory), batch_size)
states = np.array([self.memory[i][0] for i in batch])
actions = np.array([self.memory[i][1] for i in batch])
rewards = np.array([self.memory[i][2] for i in batch])
next_states = np.array([self.memory[i][3] for i in batch])
dones = np.array([self.memory[i][4] for i in batch])
target_q_values = self.model.predict(states, verbose=0)
target_q_values_next = self.target_model.predict(next_states, verbose=0)
for i in range(batch_size):
if dones[i]:
target_q_values[i, actions[i]] = rewards[i]
else:
max_next_q = np.max(target_q_values_next[i])
target_q_values[i, actions[i]] = rewards[i] + self.gamma * max_next_q
self.model.fit(states, target_q_values, epochs=1, verbose=0)
if self.epsilon > self.epsilon_min:
self.epsilon *= self.epsilon_decay
def train(self, episodes=1000):
for episode in range(episodes):
state = self.env.reset()
# Stack 4 frames
state = np.stack([self.preprocess_frame(state)] * 4)
done = False
total_reward = 0
while not done and self.frame_count < 1000000:
action = self.act(state)
next_frame, reward, done, _ = self.env.step(action)
next_frame = self.preprocess_frame(next_frame)
next_state = np.append(state[1:], [next_frame], axis=0)
self.remember(state, action, reward, next_state, done)
self.replay(32)
total_reward += reward
state = next_state
if episode % 1000 == 0:
self.update_target_model()
if episode % 100 == 0:
print(f"Episode {episode}: Reward = {total_reward}")
Project 3: Policy Gradient for Continuous Control
Environment: MuJoCo (robotics simulation)
class PolicyGradientAgent:
def __init__(self, state_size, action_size):
self.state_size = state_size
self.action_size = action_size
# Actor network (policy)
self.actor = self.build_actor()
# Critic network (value)
self.critic = self.build_critic()
def build_actor(self):
model = tf.keras.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_dim=self.state_size),
tf.keras.layers.Dense(64, activation='relu'),
tf.keras.layers.Dense(self.action_size, activation='tanh')
])
return model
def build_critic(self):
model = tf.keras.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_dim=self.state_size),
tf.keras.layers.Dense(64, activation='relu'),
tf.keras.layers.Dense(1)
])
return model
def train_episode(self, env):
state = env.reset()
states, actions, rewards = [], [], []
done = False
while not done:
# Get action from actor
action = self.actor.predict(state[np.newaxis])[0]
action = np.clip(action, -1, 1)
next_state, reward, done, _ = env.step(action)
states.append(state)
actions.append(action)
rewards.append(reward)
state = next_state
# Compute returns
returns = []
cumulative = 0
for r in reversed(rewards):
cumulative = r + 0.99 * cumulative
returns.insert(0, cumulative)
# Update critic and actor
self.update_networks(states, actions, returns)
return sum(rewards)
def update_networks(self, states, actions, returns):
# Update critic
critic_loss = tf.keras.losses.mse(
returns,
self.critic(np.array(states)).numpy().flatten()
)
# ...gradient descent...
# Update actor
# ...policy gradient update...
pass
# Usage
env = gym.make('HalfCheetah-v3')
agent = PolicyGradientAgent(env.observation_space.shape[0],
env.action_space.shape[0])
for episode in range(1000):
reward = agent.train_episode(env)
if episode % 100 == 0:
print(f"Episode {episode}: Reward = {reward:.1f}")
Summary
RL projects teach:
- Algorithm implementation
- Environment interaction
- Reward design
- Training stability
- Real-world applications