Navigating Environmental Hazards in Reinforcement Learning

Introduction

Welcome back to our course on " Navigating RL Challenges: Strategies and Future Directions "! This is the third lesson in our journey, and I'm excited to continue building on what we've learned so far. So far, we implemented random goals to create more dynamic grid world environments, followed by an exploration of reward shaping to provide more informative feedback to our agents, helping them learn more efficiently. Today, we'll tackle another critical challenge in Reinforcement Learning: environmental hazards . In real-world applications, agents often need to navigate environments with dangers or obstacles that must be avoided. Think of a robot that needs to avoid stairs while cleaning a house or an autonomous vehicle that must recognize and avoid hazards on the road. We'll implement mines in our grid world that the agent must learn to avoid, adding risk assessment and danger avoidance to our agent's learning objectives. This brings us one step closer to training agents that can handle the complexities and dangers of real-world environments.

Implementing Mines in the Environment

Let's start by modifying our grid world environment class to include mines. We'll focus first on how to initialize our environment with mines: Python def __init__(self, size=5, use_reward_shaping=True, n_mines=3): """ An extended GridWorld with: - Optional reward shaping - Optional mines that cause immediate failure """ self.size = size self.use_reward_shaping = use_reward_shaping self.n_mines = n_mines # Number of mines to place self.state = None self.goal_state = None self.mines = set() # Will store mine positions self.prev_distance = None self.action_space = 4 self.max_steps = size * 3 self.steps_count = 0 self.rng = np.random.default_rng(42) def __init__(self, size=5, use_reward_shaping=True, n_mines=3): """ An extended GridWorld with: - Optional reward shaping - Optional mines that cause immediate failure """ self.size = size self.use_reward_shaping = use_reward_shaping self.n_mines = n_mines # Number of mines to place self.state = None self.goal_state = None self.mines = set() # Will store mine positions self.prev_distance = None self.action_space = 4 self.max_steps = size * 3 self.steps_count = 0 self.rng = np.random.default_rng(42) This constructor introduces a new parameter n_mines that determines how many hazards will be placed in the environment. We also initialize an empty set self.mines that will store the locations of these mines. To visualize how mines appear in our environment, here's how a 5x5 grid would render with the agent A at (1, 1), the goal G at (4, 4), and mines * at positions (1, 2), (2, 3), and (3, 1) : text . . . . . --> safe for A to go up . A * . . --> --> safe for A to go left, but if it goes right, game over! . . . * . --> safe for A to go down . * . . . . . . . G

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Python

def __init__(self, size=5, use_reward_shaping=True, n_mines=3):
    """
    An extended GridWorld with:
    - Optional reward shaping
    - Optional mines that cause immediate failure
    """
    self.size = size
    self.use_reward_shaping = use_reward_shaping
    self.n_mines = n_mines  # Number of mines to place
    self.state = None
    self.goal_state = None
    self.mines = set()      # Will store mine positions
    self.prev_distance = None
    self.action_space = 4
    self.max_steps = size * 3
    self.steps_count = 0
    self.rng = np.random.default_rng(42)

Python

def reset(self):
    """
    Reset environment with agent at (0,0), random goal, and random mines
    """
    self.state = (0, 0)
    self.goal_state = self._sample_random_goal()
    self.mines = self._sample_mines() if self.n_mines > 0 else set()
    self.prev_distance = self._manhattan_distance(self.state, self.goal_state)
    self.steps_count = 0
    mines_near = self._check_nearby_mines()
    return (self.state, self.goal_state, mines_near)

def reset(self):
    """
    Reset environment with agent at (0,0), random goal, and random mines
    """
    self.state = (0, 0)
    self.goal_state = self._sample_random_goal()
    self.mines = self._sample_mines() if self.n_mines > 0 else set()
    self.prev_distance = self._manhattan_distance(self.state, self.goal_state)
    self.steps_count = 0
    mines_near = self._check_nearby_mines()
    return (self.state, self.goal_state, mines_near)

Python

def _sample_mines(self):
    """Randomly place mines, avoiding (0, 0) and the goal."""
    available = [(r, c) for r in range(self.size)
                for c in range(self.size)
                if (r, c) not in [(0, 0), self.goal_state]]
    
    n_placements = min(self.n_mines, len(available))
    chosen = self.rng.choice(len(available), n_placements, replace=False)
    mines = {available[i] for i in chosen}
    return mines

def _sample_mines(self):
    """Randomly place mines, avoiding (0, 0) and the goal."""
    available = [(r, c) for r in range(self.size)
                for c in range(self.size)
                if (r, c) not in [(0, 0), self.goal_state]]
    
    n_placements = min(self.n_mines, len(available))
    chosen = self.rng.choice(len(available), n_placements, replace=False)
    mines = {available[i] for i in chosen}
    return mines

Python

def _check_nearby_mines(self):
    """
    Check if there are mines in adjacent cells.
    Returns a binary tuple (up, down, left, right) where 1 indicates danger.
    """
    r, c = self.state
    nearby = [
        (max(r - 1, 0), c),             # up
        (min(r + 1, self.size - 1), c),   # down
        (r, max(c - 1, 0)),             # left
        (r, min(c + 1, self.size - 1))    # right
    ]
    return tuple(1 if pos in self.mines else 0 for pos in nearby)

def _check_nearby_mines(self):
    """
    Check if there are mines in adjacent cells.
    Returns a binary tuple (up, down, left, right) where 1 indicates danger.
    """
    r, c = self.state
    nearby = [
        (max(r - 1, 0), c),             # up
        (min(r + 1, self.size - 1), c),   # down
        (r, max(c - 1, 0)),             # left
        (r, min(c + 1, self.size - 1))    # right
    ]
    return tuple(1 if pos in self.mines else 0 for pos in nearby)

Python

# Part of the step method that handles mine collisions
if self.state in self.mines:
    mines_near = self._check_nearby_mines()
    return (self.state, self.goal_state, mines_near), -1.0, True, {"hit_mine": True}

# Part of the step method that handles mine collisions
if self.state in self.mines:
    mines_near = self._check_nearby_mines()
    return (self.state, self.goal_state, mines_near), -1.0, True, {"hit_mine": True}

Python

def run_episode(env, agent, render=False, deterministic=False):
    """Execute a single episode and return results"""
    state = env.reset()
    total_reward = 0.0
    done = False

    if render:
        print("Initial State:")
        env.render()

    while not done:
        # Agent selects action (explore or exploit)
        action = agent.act(state, deterministic=deterministic)
        
        # Environment responds with next state and reward
        next_state, reward, done, info = env.step(action)
        
        # Agent learns from the experience
        agent.learn(state, action, reward, next_state, done)
        
        state = next_state
        total_reward += reward

        if render:
            env.render()

    success = (not info.get("timeout", False)) and (not info.get("hit_mine", False))
    hit_mine = info.get("hit_mine", False)
    return total_reward, success, hit_mine

def run_episode(env, agent, render=False, deterministic=False):
    """Execute a single episode and return results"""
    state = env.reset()
    total_reward = 0.0
    done = False

    if render:
        print("Initial State:")
        env.render()

    while not done:
        # Agent selects action (explore or exploit)
        action = agent.act(state, deterministic=deterministic)
        
        # Environment responds with next state and reward
        next_state, reward, done, info = env.step(action)
        
        # Agent learns from the experience
        agent.learn(state, action, reward, next_state, done)
        
        state = next_state
        total_reward += reward

        if render:
            env.render()

    success = (not info.get("timeout", False)) and (not info.get("hit_mine", False))
    hit_mine = info.get("hit_mine", False)
    return total_reward, success, hit_mine

Python

def main():
    """Compare performance with different hazard densities"""
    env_size = 5
    actions = [0, 1, 2, 3]
    n_episodes = 1000

    # Try environments with increasing danger
    mine_counts = [0, 2, 5]

    for n_mines in mine_counts:
        print(f"\nTraining with {n_mines} mines...")
        env = GridWorldEnv(size=env_size, use_reward_shaping=True, n_mines=n_mines)
        agent = QLearningAgent(actions)
        train_agent(env, agent, n_episodes)

def main():
    """Compare performance with different hazard densities"""
    env_size = 5
    actions = [0, 1, 2, 3]
    n_episodes = 1000

    # Try environments with increasing danger
    mine_counts = [0, 2, 5]

    for n_mines in mine_counts:
        print(f"\nTraining with {n_mines} mines...")
        env = GridWorldEnv(size=env_size, use_reward_shaping=True, n_mines=n_mines)
        agent = QLearningAgent(actions)
        train_agent(env, agent, n_episodes)

Python

def reset(self):
    """
    Reset environment with agent at (0,0), random goal, and random mines
    """
    self.state = (0, 0)
    self.goal_state = self._sample_random_goal()
    self.mines = self._sample_mines() if self.n_mines > 0 else set()
    self.prev_distance = self._manhattan_distance(self.state, self.goal_state)
    self.steps_count = 0
    mines_near = self._check_nearby_mines()
    return (self.state, self.goal_state, mines_near)

Python

def _sample_mines(self):
    """Randomly place mines, avoiding (0, 0) and the goal."""
    available = [(r, c) for r in range(self.size)
                for c in range(self.size)
                if (r, c) not in [(0, 0), self.goal_state]]
    
    n_placements = min(self.n_mines, len(available))
    chosen = self.rng.choice(len(available), n_placements, replace=False)
    mines = {available[i] for i in chosen}
    return mines

Python

def _check_nearby_mines(self):
    """
    Check if there are mines in adjacent cells.
    Returns a binary tuple (up, down, left, right) where 1 indicates danger.
    """
    r, c = self.state
    nearby = [
        (max(r - 1, 0), c),             # up
        (min(r + 1, self.size - 1), c),   # down
        (r, max(c - 1, 0)),             # left
        (r, min(c + 1, self.size - 1))    # right
    ]
    return tuple(1 if pos in self.mines else 0 for pos in nearby)

Python

# Part of the step method that handles mine collisions
if self.state in self.mines:
    mines_near = self._check_nearby_mines()
    return (self.state, self.goal_state, mines_near), -1.0, True, {"hit_mine": True}

Python

def run_episode(env, agent, render=False, deterministic=False):
    """Execute a single episode and return results"""
    state = env.reset()
    total_reward = 0.0
    done = False

    if render:
        print("Initial State:")
        env.render()

    while not done:
        # Agent selects action (explore or exploit)
        action = agent.act(state, deterministic=deterministic)
        
        # Environment responds with next state and reward
        next_state, reward, done, info = env.step(action)
        
        # Agent learns from the experience
        agent.learn(state, action, reward, next_state, done)
        
        state = next_state
        total_reward += reward

        if render:
            env.render()

    success = (not info.get("timeout", False)) and (not info.get("hit_mine", False))
    hit_mine = info.get("hit_mine", False)
    return total_reward, success, hit_mine

Python

def main():
    """Compare performance with different hazard densities"""
    env_size = 5
    actions = [0, 1, 2, 3]
    n_episodes = 1000

    # Try environments with increasing danger
    mine_counts = [0, 2, 5]

    for n_mines in mine_counts:
        print(f"\nTraining with {n_mines} mines...")
        env = GridWorldEnv(size=env_size, use_reward_shaping=True, n_mines=n_mines)
        agent = QLearningAgent(actions)
        train_agent(env, agent, n_episodes)

Introduction

Implementing Mines in the Environment

Join the 1M+ learners on CodeSignal

Updating Resetting Logic

Placing Mines

Detecting Nearby Danger

Handling Mine Collisions

Training with Risk Assessment

Comparing Risk Levels

Visualizing Different Mine Counts

Conclusion and Next Steps