I am copying an introduction video to reinforcement learning and I get an error saying ValueError: "Can't convert non-rectangular Python sequence to Tensor". in the code typed at 22:11 (def choose_action) after finalizing all the code. https://www.youtube.com/watch?v=LawaN3BdI00
after some searching I found that there is a difference between tensorflow 1 and tensorflow 2 and that I should use
state = tf.ragged.constant(observation)
according to ValueError: Can't convert non-rectangular Python sequence to Tensor
the "observation" that is used as input to the tf.ragged.constant function is as such:
[array([-0.00291625, 0.00813361, 0.03261928, 0.02612764], dtype=float32)]
the output "state" is as such:
<tf.RaggedTensor [[-0.0029162522, 0.008133613, 0.032619275, 0.026127638]]>
with shape=(4,)
The video states that the state should have multiple dimensions, therefore it was put in brackets. I think I have to reshape to (4,1). I have tried the following and all sorts of combinations:
state = np.array(observation).reshape(4, 1)
state = tf.expand_dims(observation, 1)
state = tf.expand_dims([observation], 1)
state = np.array(observation).reshape(len(observation), 1)
state = tf.ragged.constant(observation)
state = tf.ragged.constant([observation])
When I use tf.ragged.constant(observation) as mentioned in the stackoverflow question, It returns an error:
The last dimension of the inputs to a Dense layer should be defined. Found None. Full input shape received: (1, None) Call arguments received by layer "actor_critic_network" " f"(type ActorCriticNetwork): • state=<tf.RaggedTensor [[-0.035476074, -0.015764749, -0.032257307, -0.04707874]]>
I have read https://www.tensorflow.org/guide/ragged_tensor but it is still not clear to me how I can solve this problem. Could someone please help me to define the outputshape of my tensor?
Minimal reproducible sample
import gym
import numpy as np
import pandas as pd
import keras
from keras import layers
import tensorflow as tf
from keras.optimizers import Adam
import tensorflow_probability as tfp
import os
seed = 43
gamma = 0.99 # discount factor for past rewards
max_steps_per_episode = 65000
class ActorCriticNetwork(keras.Model):
def __init__(self, n_actions, fc1_dims=1024, fc2_dims=512, #number of actions as input, number of fully connected layers
name='actor_critic', chkpt_dir=r'C:\User\Documents'): #name for model checkpoint purposes, and directory for checkpoint
super(ActorCriticNetwork, self).__init__()
self.fc1_dims = fc1_dims
self.fc2_dims = fc2_dims
self.n_actions = n_actions
self.model_name = name
self.checkpoint_dir = chkpt_dir
self.checkpoint_file = os.path.join(self.checkpoint_dir, name+'_ac') # dont want to confuse modeltypes
self.fc1 = layers.Dense(self.fc1_dims, activation='relu')
self.fc2 = layers.Dense(self.fc2_dims, activation='relu')
self.v = layers.Dense(1, activation=None)# two seperate outputs, two common layer and two independent outputs
self.pi = layers.Dense(n_actions, activation='softmax')
def call(self, state): # define call function -> feed forward
value = self.fc1(state)
value = self.fc2(value)
v = self.v(value) # value function
pi = self.pi(value) # policy pi
return v, pi
class Agent:
def __init__(self, alpha=0.0003, gamma=0.99, n_actions=2):
self.gamma = gamma
self.n_actions = n_actions
self.action = None
self.action_space = [i for i in range(0, self.n_actions)] # list of random selections
self.actor_critic = ActorCriticNetwork(n_actions=n_actions)
self.actor_critic.compile(optimizer=Adam(learning_rate=alpha))
def choose_action(self, observation):
# state = tf.expand_dims(observation, 1)
# state = tf.RaggedTensor.from_row_lengths(values=observation, row_lengths=[4, 1])
# state = tf.ragged.constant(observation)
# state = tf.expand_dims(observation, 1)
# state = np.array(observation).reshape(len(observation), 1)
state = tf.convert_to_tensor([observation]) # current state of environment-add an extra dimension because batches are expected
_, probs = self.actor_critic(state) # we dont care about the value of the state so we choose a blank
action_probabilities = tfp.distribution.Categorical(probs=probs)# use that output, the probablities defined by the neural network, to feed the actual tf probability distribution to select an action by sampling that
action = action_probabilities.sample() # select an action from the samples
# we dont need the log probability now, because this calc takes place outside of the gradient tape, it allows to calc gradients manually
self.action = action
return action.numpy()[0]
def save_model(self):
print('...saving model...')
self.actor_critic.save_weights(self.actor_critic.checkpoint_file)
def load_model(self):
print('...loading models...')
self.actor_critic.load_weights(self.actor_critic.checkpoint_file)
def learn(self, state, reward, state_, done):
state = tf.convert_to_tensor([state], dtype=tf.float32)
state_ = tf.convert_to_tensor([state_], dtype=tf.float32)
reward = tf.convert_to_tensor(reward, dtype=tf.float32) # we dont have to add a batch dimension to the reward beacuse it is not fed to a deep nn
with tf.GradientTape() as tape: #dont know if it is needed
state_value, probs = self.actor_critic(state) #feed our current state to the actorcriticnetwork and get back our quantities of interest
state_value_, _ = self.actor_critic(state_) #feed our new state to the actorcriticnetwork and get back our quantities of interest, for the new state we dont care about the probabilities but just the values
state_value = tf.squeeze(state_value)# for the calculation of the loss we have to get rid of the extra dimension; the loss works best on a 1 dimensional quantity
state_value_ = tf.squeeze(state_value_)
action_probs = tfp.distributions.Categorical(probs=probs) # define probs by the output of our deep neural network
log_prob = action_probs.log_prob(self.action) #this is the log prob that we saved for the actions of the agent
delta = reward + self.gamma*state_value_*(1-int(done)) - state_value #
actor_loss = -log_prob*delta
critic_loss = delta**2
total_loss = actor_loss + critic_loss
gradient = tape.gradient(total_loss, self.actor.critic.trainable_variables)
self.actor_critic.optimizer.apply_gradients(zip(
gradient, self.actor_critic.trainable_variables))
if __name__ == '__main__':
env = gym.make('CartPole-v1')
agent = Agent(alpha=1e-5, n_actions=env.action_space.n)
n_games=1000
filename = 'cartpole.png'
figure_file = 'plots/'+filename
best_score =env.reward_range[0]
score_history = []
load_checkpoint = False
if load_checkpoint:
agent.load_model()
for i in range(n_games):
observation = env.reset()
done = False
score = 0
while not done:
action = agent.choose_action(observation)
observation_, reward, done, info = env.step(action)
score += reward
if not load_checkpoint:
agent.learn(observation, reward, observation_, done)
observation = observation_
score_history.append(score)
avg_score = np.mean(score_history[-100:])
if avg_score > best_score:
best_score = avg_score
if not load_checkpoint:
agent.save_models()
print('episode', i, 'score%.1f' % score, 'avg_score %.1f' % avg_score)
x = [i+1 for i in range(n_games)]