pythonnicalmanrl 1.0.3
Sort implementation in Python

# Q LEARNING 
import gym
import numpy as np

env = gym.make('FrozenLake-v1')
q_table = np.zeros((env.observation_space.n, env.action_space.n))

num_episodes = 10000
max_steps_per_episode = 100
lr = 0.1
dr = 0.99
er = 1
max_er = 1
min_er = 0.01
edr = 0.001

for episode in range(num_episodes):
    state = env.reset()
    done = False

    for step in range(max_steps_per_episode):
        explore_threshold = np.random.uniform(0, 1)
        if explore_threshold > er:
            action = np.argmax(q_table[state, :])
        else:
            action = env.action_space.sample()

        new_state, reward, done, _ = env.step(action)

        q_table[state, action] = q_table[state, action] * (1 - lr) + \
                                 lr * (reward + dr * np.max(q_table[new_state, :]))

        state = new_state

        if done:
            break

    er = min_er + (max_er - min_er) * np.exp(-edr * episode)

print("Q-table:")
print(q_table)



# SARSA

import gym
import numpy as np

env = gym.make('FrozenLake-v1')
q_table = np.zeros((env.observation_space.n, env.action_space.n))

num_episodes = 10000
max_steps_per_episode = 100
learning_rate = 0.1
discount_rate = 0.99
exploration_rate = 1
max_exploration_rate = 1
min_exploration_rate = 0.01
exploration_decay_rate = 0.001

for episode in range(num_episodes):
    state = env.reset()
    done = False

    # Exploration-exploitation trade-off
    exploration_rate_threshold = np.random.uniform(0, 1)
    if exploration_rate_threshold > exploration_rate:
        action = np.argmax(q_table[state, :])
    else:
        action = env.action_space.sample()

    for step in range(max_steps_per_episode):
        new_state, reward, done, _ = env.step(action)

        # Exploration-exploitation trade-off
        exploration_rate_threshold = np.random.uniform(0, 1)
        if exploration_rate_threshold > exploration_rate:
            new_action = np.argmax(q_table[new_state, :])
        else:
            new_action = env.action_space.sample()

        # Update Q-table
        q_table[state, action] = q_table[state, action] + learning_rate * (
                reward + discount_rate * q_table[new_state, new_action] - q_table[state, action])

        state = new_state
        action = new_action

        if done:
            break

    # Exploration rate decay
    exploration_rate = min_exploration_rate + \
                       (max_exploration_rate - min_exploration_rate) * np.exp(-exploration_decay_rate * episode)

print("Q-table:")
print(q_table)

# AVI

import numpy as np

class GridWorld:
    def __init__(self):
        self.num_states = 25
        self.num_actions = 4
        self.grid_size = 5
        self.terminal_state = 24

    def reward_function(self, state, action, next_state):
        if next_state == self.terminal_state:
            return 10
        else:
            return 0

def approximate_value_iteration(env, gamma, epsilon, max_iterations):
    V = np.zeros(env.num_states)

    for iteration in range(max_iterations):
        delta = 0
        
        for state in range(env.num_states):
            q_values = []
            for action in range(env.num_actions):
                next_state_rewards = []
                for next_state in range(env.num_states):
                    reward = env.reward_function(state, action, next_state)
                    next_state_rewards.append(reward + gamma * V[next_state])
                
                q_value = np.mean(next_state_rewards)
                q_values.append(q_value)
            
            new_value = max(q_values)
            delta = max(delta, np.abs(new_value - V[state]))
            V[state] = new_value
        
        if delta < epsilon:
            break
    
    return V

env = GridWorld()
gamma = 0.9
epsilon = 0.01
max_iterations = 1000

V = approximate_value_iteration(env, gamma, epsilon, max_iterations)

print("Optimal Value Function:")
print(V.reshape(env.grid_size, env.grid_size))

# API

import numpy as np

class GridWorld:
    def __init__(self):
        self.num_states = 16
        self.num_actions = 4
        self.grid_size = 4
        self.terminal_state = 15
        self.policy = np.zeros(self.num_states, dtype=int)

    def reward_function(self, state, action, next_state):
        if next_state == self.terminal_state:
            return 10
        else:
            return -1

    def transition_probability(self, state, action, next_state):
        if state == next_state:
            return 0.0

        if action == 0:  # Up
            if next_state == state - self.grid_size:
                return 0.8
            elif next_state == state + 1 or next_state == state - 1:
                return 0.1
            else:
                return 0.0
        elif action == 1:  # Down
            if next_state == state + self.grid_size:
                return 0.8
            elif next_state == state + 1 or next_state == state - 1:
                return 0.1
            else:
                return 0.0
        elif action == 2:  # Left
            if next_state == state - 1:
                return 0.8
            elif next_state == state - self.grid_size or next_state == state + self.grid_size:
                return 0.1
            else:
                return 0.0
        elif action == 3:  # Right
            if next_state == state + 1:
                return 0.8
            elif next_state == state - self.grid_size or next_state == state + self.grid_size:
                return 0.1
            else:
                return 0.0

def approximate_policy_iteration(env, gamma, epsilon, max_iterations):
    V = np.zeros(env.num_states)

    for iteration in range(max_iterations):
        # Policy Evaluation
        delta = 0
        
        for state in range(env.num_states):
            old_value = V[state]
            q_value = 0
            
            for next_state in range(env.num_states):
                reward = env.reward_function(state, env.policy[state], next_state)
                q_value += env.transition_probability(state, env.policy[state], next_state) * (reward + gamma * V[next_state])
            
            V[state] = q_value
            delta = max(delta, np.abs(old_value - V[state]))
        
        # Policy Improvement
        policy_stable = True
        
        for state in range(env.num_states):
            old_action = env.policy[state]
            q_values = []
            
            for action in range(env.num_actions):
                q_value = 0
                for next_state in range(env.num_states):
                    reward = env.reward_function(state, action, next_state)
                    q_value += env.transition_probability(state, action, next_state) * (reward + gamma * V[next_state])
                
                q_values.append(q_value)
            
            env.policy[state] = np.argmax(q_values)
            
            if old_action != env.policy[state]:
                policy_stable = False
        
        if delta < epsilon and policy_stable:
            break
    
    return V, env.policy

env = GridWorld()
gamma = 0.9
epsilon = 0.01
max_iterations = 1000

V, policy = approximate_policy_iteration(env, gamma, epsilon, max_iterations)

print("Optimal Value Function:")
print(V.reshape(env.grid_size, env.grid_size))

print("Optimal Policy:")
print(policy.reshape(env.grid_size, env.grid_size))


# ROLLOUT

import random

def rollout(state, policy, max_depth):
    reward = 0
    for _ in range(max_depth):
        action = policy(state)
        state, r = take_action(state, action)
        reward += r
    return reward

def take_action(state, action):
    # Implement your environment's logic for taking an action
    # and returning the next state and the reward
    # Here's a placeholder implementation for demonstration purposes
    if action == 'left':
        return state - 1, random.randint(-10, 10)
    elif action == 'right':
        return state + 1, random.randint(-10, 10)
    else:
        return state, random.randint(-10, 10)

# Define your policy function
def policy(state):
    # Implement your policy logic here
    # This is a placeholder policy that selects a random action
    return random.choice(['left', 'right'])

# Test the rollout algorithm
initial_state = 0
max_depth = 10
num_rollouts = 100

total_reward = 0
for _ in range(num_rollouts):
    total_reward += rollout(initial_state, policy, max_depth)

average_reward = total_reward / num_rollouts
print("Average reward:", average_reward)


# RTDP
import random

# Define the states, actions, and initial state for the grid-world problem
states = [(0, 0), (0, 1), (0, 2), (1, 0), (1, 1), (1, 2), (2, 0), (2, 1), (2, 2)]
actions = {
    (0, 0): {'right': (0, 1), 'down': (1, 0)},
    (0, 1): {'left': (0, 0), 'right': (0, 2), 'down': (1, 1)},
    (0, 2): {'left': (0, 1), 'down': (1, 2)},
    (1, 0): {'up': (0, 0), 'right': (1, 1), 'down': (2, 0)},
    (1, 1): {'up': (0, 1), 'left': (1, 0), 'right': (1, 2), 'down': (2, 1)},
    (1, 2): {'up': (0, 2), 'left': (1, 1), 'down': (2, 2)},
    (2, 0): {'up': (1, 0), 'right': (2, 1)},
    (2, 1): {'up': (1, 1), 'left': (2, 0), 'right': (2, 2)},
    (2, 2): {'up': (1, 2), 'left': (2, 1)}
}
initial_state = (0, 0)
discount_factor = 0.9
max_iterations = 100

# Function to simulate the environment and get the next state and reward
def simulate_environment(state, action):
    # Simulate a random environment
    next_state = actions[state][action]
    if next_state == (2, 2):  # Goal state
        reward = 1
    elif next_state in [(0, 1), (1, 2)]:  # Cliff states
        reward = -1
    else:
        reward = 0
    return next_state, reward

# Run the RTDP algorithm
def rtdp(states, actions, initial_state, discount_factor, max_iterations):
    # Initialize the value function
    value_function = {state: 0.0 for state in states}

    # Iterate until convergence or maximum iterations reached
    for i in range(max_iterations):
        state = initial_state

        # Loop until a terminal state is reached
        while state not in [(2, 2), (0, 1), (1, 2)]:
            # Select an action based on the value function
            action = max(actions[state], key=lambda a: value_function[actions[state][a]])

            # Simulate the environment to get the next state and reward
            next_state, reward = simulate_environment(state, action)

            # Update the value function using the Bellman equation
            value_function[state] += discount_factor * (reward + value_function[next_state] - value_function[state])

            # Move to the next state
            state = next_state

    return value_function

# Run the RTDP algorithm for the grid-world problem
value_function = rtdp(states, actions, initial_state, discount_factor, max_iterations)

# Print the value function
for state in states:
    print(f"State: {state} - Value: {value_function[state]}")
    

# ACTOR CRITIC

import gym
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# Configuration parameters for the whole setup
seed = 42
gamma = 0.99  # Discount factor for past rewards
max_steps_per_episode = 10000
env = gym.make("CartPole-v0")  # Create the environment
env.seed(seed)
eps = np.finfo(np.float32).eps.item()  # Smallest number such that 1.0 + eps != 1.0

num_inputs = 4
num_actions = 2
num_hidden = 128

inputs = layers.Input(shape=(num_inputs,))
common = layers.Dense(num_hidden, activation="relu")(inputs)
action = layers.Dense(num_actions, activation="softmax")(common)
critic = layers.Dense(1)(common)

model = keras.Model(inputs=inputs, outputs=[action, critic])

optimizer = keras.optimizers.Adam(learning_rate=0.01)
huber_loss = keras.losses.Huber()
action_probs_history = []
critic_value_history = []
rewards_history = []
running_reward = 0
episode_count = 0

while True:  # Run until solved
    state = env.reset()
    episode_reward = 0
    with tf.GradientTape() as tape:
        for timestep in range(1, max_steps_per_episode):
            # env.render(); Adding this line would show the attempts
            # of the agent in a pop up window.

            state = tf.convert_to_tensor(state)
            state = tf.expand_dims(state, 0)

            # Predict action probabilities and estimated future rewards
            # from environment state
            action_probs, critic_value = model(state)
            critic_value_history.append(critic_value[0, 0])

            # Sample action from action probability distribution
            action = np.random.choice(num_actions, p=np.squeeze(action_probs))
            action_probs_history.append(tf.math.log(action_probs[0, action]))

            # Apply the sampled action in our environment
            state, reward, done, _ = env.step(action)
            rewards_history.append(reward)
            episode_reward += reward

            if done:
                break

        # Update running reward to check condition for solving
        running_reward = 0.05 * episode_reward + (1 - 0.05) * running_reward

        # Calculate expected value from rewards
        # - At each timestep what was the total reward received after that timestep
        # - Rewards in the past are discounted by multiplying them with gamma
        # - These are the labels for our critic
        returns = []
        discounted_sum = 0
        for r in rewards_history[::-1]:
            discounted_sum = r + gamma * discounted_sum
            returns.insert(0, discounted_sum)

        # Normalize
        returns = np.array(returns)
        returns = (returns - np.mean(returns)) / (np.std(returns) + eps)
        returns = returns.tolist()

        # Calculating loss values to update our network
        history = zip(action_probs_history, critic_value_history, returns)
        actor_losses = []
        critic_losses = []
        for log_prob, value, ret in history:
            # At this point in history, the critic estimated that we would get a
            # total reward = `value` in the future. We took an action with log probability
            # of `log_prob` and ended up recieving a total reward = `ret`.
            # The actor must be updated so that it predicts an action that leads to
            # high rewards (compared to critic's estimate) with high probability.
            diff = ret - value
            actor_losses.append(-log_prob * diff)  # actor loss

            # The critic must be updated so that it predicts a better estimate of
            # the future rewards.
            critic_losses.append(
                huber_loss(tf.expand_dims(value, 0), tf.expand_dims(ret, 0))
            )

        # Backpropagation
        loss_value = sum(actor_losses) + sum(critic_losses)
        grads = tape.gradient(loss_value, model.trainable_variables)
        optimizer.apply_gradients(zip(grads, model.trainable_variables))

        # Clear the loss and reward history
        action_probs_history.clear()
        critic_value_history.clear()
        rewards_history.clear()

    # Log details
    episode_count += 1
    if episode_count % 10 == 0:
        template = "running reward: {:.2f} at episode {}"
        print(template.format(running_reward, episode_count))

    if running_reward > 195:  # Condition to consider the task solved
        print("Solved at episode {}!".format(episode_count))
        break
        
# DOCUMENTATION OF PROGRAMS