Elevator Q-Learning Pipeline

A step-by-step guide to applying Q-learning to control a simple multi-floor elevator to minimize wait time.

Libraries Used

This project uses a few key Python libraries. Here is a quick overview of their roles:

• numpy: Used for numerical operations and managing arrays.
• random: Used for generating random numbers for the simulation.
• collections: Used for the `defaultdict` to store the Q-table.

1. Define the Environment

• What the code does: It defines an `ElevatorEnv` class that simulates an elevator with a few floors and waiting passengers.
• Behind the scenes: The environment's state includes the elevator's current floor and the floors where people are waiting. The actions are moving up, moving down, or staying put. The reward is a negative value for each passenger's waiting time, encouraging the agent to move quickly.
• Abstraction: The `ElevatorEnv` class abstracts the real-world problem into a simple, interactive environment for the agent.

import numpy as np
import random

class ElevatorEnv:
    def __init__(self, n_floors=4):
        self.n_floors = n_floors
        self.state = {"floor": 0, "passengers_waiting": set()}

    def reset(self):
        self.state["floor"] = 0
        self.state["passengers_waiting"] = set(random.sample(range(self.n_floors), k=1))
        return tuple(sorted(list(self.state["passengers_waiting"])) + [self.state["floor"]])

    def step(self, action):
        # Action 0: move down, 1: stay, 2: move up
        if action == 0 and self.state["floor"] > 0:
            self.state["floor"] -= 1
        elif action == 2 and self.state["floor"] < self.n_floors - 1:
            self.state["floor"] += 1

        # Drop off passengers
        if self.state["floor"] in self.state["passengers_waiting"]:
            self.state["passengers_waiting"].remove(self.state["floor"])
        
        # New passenger might appear
        if random.random() < 0.2:
            new_floor = random.randint(0, self.n_floors - 1)
            self.state["passengers_waiting"].add(new_floor)

        reward = -len(self.state["passengers_waiting"]) # Negative reward for waiting passengers
        done = len(self.state["passengers_waiting"]) == 0

        return tuple(sorted(list(self.state["passengers_waiting"])) + [self.state["floor"]]), reward, done

2. Implement the Q-learning Agent

• What the code does: It defines a `QLearningAgent` class that learns the optimal policy through trial and error.
• Behind the scenes: The agent maintains a "Q-table" (a `defaultdict` in this case) that stores the expected future reward for each state-action pair. It uses an epsilon-greedy strategy for action selection and updates the Q-table using the Bellman equation in the `learn` method.
• Abstraction: The `QLearningAgent` class abstracts the core logic of a model-free reinforcement learning algorithm.

from collections import defaultdict

class QLearningAgent:
    def __init__(self, n_actions, alpha=0.1, gamma=0.9, epsilon=1.0, epsilon_decay=0.995):
        self.n_actions = n_actions
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_decay = epsilon_decay
        self.Q = defaultdict(lambda: np.zeros(self.n_actions, dtype=np.float32))

    def act(self, state):
        if np.random.rand() < self.epsilon:
            return np.random.randint(self.n_actions)
        return int(np.argmax(self.Q[state]))

    def learn(self, s, a, r, s_next, done):
        best_next_q = 0.0 if done else np.max(self.Q[s_next])
        td_target = r + self.gamma * best_next_q
        td_error = td_target - self.Q[s][a]
        self.Q[s][a] += self.alpha * td_error
        
        if self.epsilon > 0.05:
            self.epsilon *= self.epsilon_decay

3. Run the Experiment

• What the code does: It runs a training loop for thousands of episodes, allowing the agent to learn the optimal policy.
• Behind the scenes: In each episode, the agent interacts with the environment, taking actions and learning from the resulting rewards and next states. Over time, its Q-table converges, and the agent's performance improves.
• Abstraction: The training loop abstracts the complex, long-term process of learning and policy improvement.

env = ElevatorEnv(n_floors=4)
agent = QLearningAgent(n_actions=3)

n_episodes = 2000
for i in range(n_episodes):
    state = env.reset()
    done = False
    episode_reward = 0
    while not done:
        action = agent.act(state)
        next_state, reward, done = env.step(action)
        agent.learn(state, action, reward, next_state, done)
        state = next_state
        episode_reward += reward
    
    if (i + 1) % 100 == 0:
        print(f"Episode {i + 1}, Epsilon: {agent.epsilon:.4f}, Total Reward: {episode_reward:.2f}")

print("\nTraining finished. You can now test the policy.")