0
$\begingroup$

I am trying to implement a Q-learning algorithm in Python for a 3D gird world as part of an assignment, wherein the environment.py file defines the actions to be taken. So far I have tried several times, but not once do the values of the reward function in any of the iterations change. This is my latest version of the code.

from collections import defaultdict
import numpy as np
import argparse
import random

#Libraries for plotting
import matplotlib
import matplotlib.style
import plotting

from environment import TreasureCube

matplotlib.style.use('ggplot')

class RandomAgent(object):
    def __init__(self):
        self.action_space = ['left','right','forward','backward','up','down'] # in TreasureCube
        self.Q = defaultdict(lambda: np.zeros(len(self.action_space)))  #Q is an array of n zeroes (here n=6).

    def take_action(self, state):
        action = random.choice(self.action_space)
        return action

    def train(self, state, action, next_state, reward):
        pass

class Agent:
    def __init__ (self):
        self.action_space = ['left','right','forward','backward','up','down']
        self.n = len(self.action_space)
        self.Q = defaultdict(lambda: np.zeros(self.n))
        self.epsilon = 0.01
        self.alpha = 0.5
        self.gamma = 0.99

    def epsilon_greedy_policy(self, Q, epsilon, n):
        def policy_function(state):
            #Selects a random action with epsilon probability out of all available actions.
            action_probabilities = np.ones(self.n, dtype=float) * self.epsilon / self.n
            
            #Selects the greedy action with (1 - epsilon) probability.
            best_action = np.argmax(self.Q[state]) 
            action_probabilities[best_action] += (1.0 - self.epsilon)
            
            return action_probabilities
        return policy_function

    def take_action(self, state, action_probabilities):
        if (np.random.random_sample() > self.epsilon):  #np.random.random_sample is an alias for np.random.random
            return np.argmax(self.Q[state])    #Selects the greedy action with (1 - epsilon) probability.
        else:
            return np.random.choice(np.arange(self.n))#, p=self.epsilon_greedy_policy(state))  #Selects a random action.

    def train(self, state, action, next_state, reward):
        #Store the existing Q(s, a) values. (NOT NEEDED)
        q_old = self.Q[state][action]
        best_next_action = np.argmax(self.Q[next_state])

        #Retrieve the expected value for next state (or 0 if no next state, i.e., end of episode)
        q_next = self.Q[next_state][best_next_action] if next_state is not None else 0

        #Update the Q-table.
        self.Q[state][action] += self.alpha * ((reward + self.gamma * q_next) - q_old)

        #print(self.Q[state][action])

def test_cube(max_episode, max_step, gamma = 0.99, alpha = 0.5, epsilon = 0.01, n = 6):
    env = TreasureCube(max_step=max_step)
    agent = Agent()
    stats = plotting.EpisodeStats(episode_lengths = np.zeros(n), episode_rewards = np.zeros(n)) 
    policy = agent.epsilon_greedy_policy(agent.Q, epsilon, n) #Declaring the policy before the loop.

    for episode_num in range(0, max_episode):
        state = env.reset()
        terminate = False
        t = 0
        episode_reward = 0
        while not terminate:
            action_probabilities = policy(state)
            action = agent.take_action(state, action_probabilities)
            reward, terminate, next_state = env.step(action)
            episode_reward += reward
            stats.episode_rewards[episode_num] = episode_reward
            #you can comment the following two lines, if the output is too much
            #env.render() # comment
            #print(f'step: {t}, action: {action}, reward: {reward}') # comment
            agent.train(state, action, next_state, reward)
            state = next_state
            t += 1
            stats.episode_lengths[episode_num] = t

        print(f'episode: {episode_num}, total_steps: {t} episode reward: {episode_reward}')

    return stats

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Test')
    parser.add_argument('--max_episode', type=int, default=500)
    parser.add_argument('--max_step', type=int, default=500)
    args = parser.parse_args()

    test_cube(args.max_episode, args.max_step)
    #plotting.plot_episode_stats(stats) 

    #Code to run: python test3.py --max_episode 50 --max_step 50

The environment.py file is as follows:

import numpy as np
import time
import random

from abc import ABC, abstractmethod

class AbstractEnvironment(ABC):
    def __init__(self):
        self.agent_sign = '+'
        self.goal_sign = 'G'
        self.corridor_sign = '-'

    def render(self):
        raise NotImplemented

    def reset(self):
        raise NotImplemented

    def step(self, action):
        raise NotImplemented

class TreasureCube(AbstractEnvironment):
    def __init__(self, max_step=20):
        super(TreasureCube, self).__init__()
        self.dim = 4
        self.max_step = max_step
        self.curr_pos = [0, 0, 0]  # (z, x, y)
        self.time_step = 0
        self.end_pos = [self.dim - 1, self.dim - 1, self.dim - 1]
        self.visual_state = []
        self.seed = None
        self.set_seed()
        self.all_actions = ['right', 'left', 'up', 'down', 'forward', 'backward']
        self.slip_actions = dict()
        self.set_action_rules()

    def reset(self):
        self.curr_pos = [0, 0, 0]
        self.time_step = 0
        self.end_pos = [self.dim - 1, self.dim - 1, self.dim - 1]
        self._reset_visual_states(self.curr_pos, self.end_pos)
        return ''.join(str(pos) for pos in self.curr_pos)

    def step(self, action, stochastic=True):
        in_action = action  # action from agent
        #assert action in self.all_actions
        reward = -0.1
        is_terminate = False
        #pre_pos = self.curr_pos
        r = random.random()
        if action == 'right':
            if r < 0.1:
                action = 'up'
            elif r < 0.2:
                action = 'down'
            elif r < 0.3:
                action = 'forward'
            elif r < 0.4:
                action = 'backward'
            else:
                action = 'right'
        elif action == 'left':
            if r < 0.1:
                action = 'up'
            elif r < 0.2:
                action = 'down'
            elif r < 0.3:
                action = 'forward'
            elif r < 0.4:
                action = 'backward'
            else:
                action = 'left'
        elif action == 'up':
            if r < 0.1:
                action = 'left'
            elif r < 0.2:
                action = 'right'
            elif r < 0.3:
                action = 'forward'
            elif r < 0.4:
                action = 'backward'
            else:
                action = 'up'
        elif action == 'down':
            if r < 0.1:
                action = 'left'
            elif r < 0.2:
                action = 'right'
            elif r < 0.3:
                action = 'forward'
            elif r < 0.4:
                action = 'backward'
            else:
                action = 'down'
        elif action == 'forward':
            if r < 0.1:
                action = 'left'
            elif r < 0.2:
                action = 'right'
            elif r < 0.3:
                action = 'up'
            elif r < 0.4:
                action = 'down'
            else:
                action = 'forward'
        else:
            if r < 0.1:
                action = 'left'
            elif r < 0.2:
                action = 'right'
            elif r < 0.3:
                action = 'up'
            elif r < 0.4:
                action = 'down'
            else:
                action = 'backward'

        if not stochastic:
            action = in_action

        assert action in self.all_actions
        if action == 'left':
            if self.curr_pos[1] == 0:  # wall
                pass
            else:
                self.curr_pos[1] -= 1
        elif action == 'right':
            if self.curr_pos[1] == self.dim - 1:  # wall
                pass
            elif self.curr_pos[1] == self.dim - 2 and self.curr_pos[0] == self.dim - 1 and self.curr_pos[2] == self.dim - 1:
                self.curr_pos[1] += 1
                is_terminate = True
                reward = 1
            else:
                self.curr_pos[1] += 1

        elif action == 'forward':
            if self.curr_pos[0] == self.dim - 1:  # wall
                pass
            elif self.curr_pos[0] == self.dim - 2 and self.curr_pos[1] == self.dim - 1 and self.curr_pos[2] == self.dim - 1:
                self.curr_pos[0] += 1
                is_terminate = True
                reward = 1
            else:
                self.curr_pos[0] += 1
        elif action == 'backward':
            if self.curr_pos[0] == 0:  # wall
                pass
            else:
                self.curr_pos[0] -= 1

        elif action == 'up':
            if self.curr_pos[2] == self.dim - 1:  # wall
                pass
            elif self.curr_pos[2] == self.dim - 2 and self.curr_pos[0] == self.dim - 1 and self.curr_pos[1] == self.dim - 1:
                self.curr_pos[2] += 1
                is_terminate = True
                reward = 1
            else:
                self.curr_pos[2] += 1
        elif action == 'down':
            if self.curr_pos[2] == 0:
                pass
            else:
                self.curr_pos[2] -= 1

        assert action in self.all_actions
        self.time_step += 1
        if self.time_step == self.max_step - 1:
            is_terminate = True

        self._reset_visual_states(self.curr_pos, self.end_pos)
        return reward, is_terminate, ''.join(str(pos) for pos in self.curr_pos)

    def render(self):
        print(' '.join(['*'] * self.dim))
        for i in range(self.dim):
            for line in self.visual_state[i]:
                print(' '.join(line))
            print(' '.join(['#'] * self.dim))
        print(' '.join(['*'] * self.dim))

    def set_seed(self, seed=10086):
        self.seed = seed
        random.seed(seed)

    def _reset_visual_states(self, agent_pos, goal_pos):
        self.visual_state = [[[self.corridor_sign] * self.dim for _ in range(self.dim)] for _ in range(self.dim)]
        self.visual_state[agent_pos[0]][agent_pos[1]][agent_pos[2]] = self.agent_sign
        self.visual_state[goal_pos[0]][goal_pos[1]][goal_pos[2]] = self.goal_sign

    def set_action_rules(self):
        self.slip_actions['right'] = ['up', 'down', 'forward', 'backward', 'right']
        self.slip_actions['left'] = ['up', 'down', 'forward', 'backward', 'left']
        self.slip_actions['up'] = ['left', 'right', 'forward', 'backward', 'up']
        self.slip_actions['down'] = ['left', 'right', 'forward', 'backward', 'down']
        self.slip_actions['forward'] = ['left', 'right', 'up', 'down', 'forward']
        self.slip_actions['backward'] = ['left', 'right', 'up', 'down', 'backward']

The output is as follows. With 50 iterations, -4.8999..., with 100 iterations, -9.8999..., and so on. This is where I do not know what to fix. I am sure I am going wrong somewhere, but where I do not know.

python test3.py --max_episode 50 --max_step 50
episode: 0, total_steps: 49 episode reward: -4.899999999999999
episode: 1, total_steps: 49 episode reward: -4.899999999999999
episode: 2, total_steps: 49 episode reward: -4.899999999999999
episode: 3, total_steps: 49 episode reward: -4.899999999999999
episode: 4, total_steps: 49 episode reward: -4.899999999999999
...
episode: 45, total_steps: 49 episode reward: -4.899999999999999
episode: 46, total_steps: 49 episode reward: -4.899999999999999
episode: 47, total_steps: 49 episode reward: -4.899999999999999
episode: 48, total_steps: 49 episode reward: -4.899999999999999
episode: 49, total_steps: 49 episode reward: -4.899999999999999
````
$\endgroup$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.