当前位置:   article > 正文

7个流行的强化学习算法及代码实现

qlearning有轮次数吗

作者:Siddhartha Pramanik

来源:DeepHub IMBA


目前流行的强化学习算法包括 Q-learning、SARSA、DDPG、A2C、PPO、DQN 和 TRPO。这些算法已被用于在游戏、机器人和决策制定等各种应用中,并且这些流行的算法还在不断发展和改进,本文我们将对其做一个简单的介绍。

fa95891d34603342224e9238deedf3a8.png


1、Q-learning

Q-learning:Q-learning 是一种无模型、非策略的强化学习算法。它使用 Bellman 方程估计最佳动作值函数,该方程迭代地更新给定状态动作对的估计值。Q-learning 以其简单性和处理大型连续状态空间的能力而闻名。

下面是一个使用 Python 实现 Q-learning 的简单示例:

  1. import numpy as np
  2. # Define the Q-table and the learning rate
  3. Q = np.zeros((state_space_size, action_space_size))
  4. alpha = 0.1
  5. # Define the exploration rate and discount factor
  6. epsilon = 0.1
  7. gamma = 0.99
  8. for episode in range(num_episodes):
  9. current_state = initial_state
  10. while not done:
  11. # Choose an action using an epsilon-greedy policy
  12. if np.random.uniform(0, 1) < epsilon:
  13. action = np.random.randint(0, action_space_size)
  14. else:
  15. action = np.argmax(Q[current_state])
  16. # Take the action and observe the next state and reward
  17. next_state, reward, done = take_action(current_state, action)
  18. # Update the Q-table using the Bellman equation
  19. Q[current_state, action] = Q[current_state, action] + alpha * (reward + gamma * np.max(Q[next_state]) - Q[current_state, action])
  20. current_state = next_state

上面的示例中,state_space_size 和 action_space_size 分别是环境中的状态数和动作数。num_episodes 是要为运行算法的轮次数。initial_state 是环境的起始状态。take_action(current_state, action) 是一个函数,它将当前状态和一个动作作为输入,并返回下一个状态、奖励和一个指示轮次是否完成的布尔值。

在 while 循环中,使用 epsilon-greedy 策略根据当前状态选择一个动作。使用概率 epsilon选择一个随机动作,使用概率 1-epsilon选择对当前状态具有最高 Q 值的动作。

采取行动后,观察下一个状态和奖励,使用Bellman方程更新q。并将当前状态更新为下一个状态。这只是 Q-learning 的一个简单示例,并未考虑 Q-table 的初始化和要解决的问题的具体细节。


2、SARSA

SARSA:SARSA 是一种无模型、基于策略的强化学习算法。它也使用Bellman方程来估计动作价值函数,但它是基于下一个动作的期望值,而不是像 Q-learning 中的最优动作。SARSA 以其处理随机动力学问题的能力而闻名。

 
 
  1. import numpy as np
  2. # Define the Q-table and the learning rate
  3. Q = np.zeros((state_space_size, action_space_size))
  4. alpha = 0.1
  5. # Define the exploration rate and discount factor
  6. epsilon = 0.1
  7. gamma = 0.99
  8. for episode in range(num_episodes):
  9. current_state = initial_state
  10. action = epsilon_greedy_policy(epsilon, Q, current_state)
  11. while not done:
  12. # Take the action and observe the next state and reward
  13. next_state, reward, done = take_action(current_state, action)
  14. # Choose next action using epsilon-greedy policy
  15. next_action = epsilon_greedy_policy(epsilon, Q, next_state)
  16. # Update the Q-table using the Bellman equation
  17. Q[current_state, action] = Q[current_state, action] + alpha * (reward + gamma * Q[next_state, next_action] - Q[current_state, action])
  18. current_state = next_state
  19. action = next_action

state_space_size和action_space_size分别是环境中的状态和操作的数量。num_episodes是您想要运行SARSA算法的轮次数。Initial_state是环境的初始状态。take_action(current_state, action)是一个将当前状态和作为操作输入的函数,并返回下一个状态、奖励和一个指示情节是否完成的布尔值。

在while循环中,使用在单独的函数epsilon_greedy_policy(epsilon, Q, current_state)中定义的epsilon-greedy策略来根据当前状态选择操作。使用概率 epsilon选择一个随机动作,使用概率 1-epsilon对当前状态具有最高 Q 值的动作。

上面与Q-learning相同,但是采取了一个行动后,在观察下一个状态和奖励时它然后使用贪心策略选择下一个行动。并使用Bellman方程更新q表。


3、DDPG

DDPG 是一种用于连续动作空间的无模型、非策略算法。它是一种actor-critic算法,其中actor网络用于选择动作,而critic网络用于评估动作。DDPG 对于机器人控制和其他连续控制任务特别有用。

 
 
  1. import numpy as np
  2. from keras.models import Model, Sequential
  3. from keras.layers import Dense, Input
  4. from keras.optimizers import Adam
  5. # Define the actor and critic models
  6. actor = Sequential()
  7. actor.add(Dense(32, input_dim=state_space_size, activation='relu'))
  8. actor.add(Dense(32, activation='relu'))
  9. actor.add(Dense(action_space_size, activation='tanh'))
  10. actor.compile(loss='mse', optimizer=Adam(lr=0.001))
  11. critic = Sequential()
  12. critic.add(Dense(32, input_dim=state_space_size, activation='relu'))
  13. critic.add(Dense(32, activation='relu'))
  14. critic.add(Dense(1, activation='linear'))
  15. critic.compile(loss='mse', optimizer=Adam(lr=0.001))
  16. # Define the replay buffer
  17. replay_buffer = []
  18. # Define the exploration noise
  19. exploration_noise = OrnsteinUhlenbeckProcess(size=action_space_size, theta=0.15, mu=0, sigma=0.2)
  20. for episode in range(num_episodes):
  21. current_state = initial_state
  22. while not done:
  23. # Select an action using the actor model and add exploration noise
  24. action = actor.predict(current_state)[0] + exploration_noise.sample()
  25. action = np.clip(action, -1, 1)
  26. # Take the action and observe the next state and reward
  27. next_state, reward, done = take_action(current_state, action)
  28. # Add the experience to the replay buffer
  29. replay_buffer.append((current_state, action, reward, next_state, done))
  30. # Sample a batch of experiences from the replay buffer
  31. batch = sample(replay_buffer, batch_size)
  32. # Update the critic model
  33. states = np.array([x[0] for x in batch])
  34. actions = np.array([x[1] for x in batch])
  35. rewards = np.array([x[2] for x in batch])
  36. next_states = np.array([x[3] for x in batch])
  37. target_q_values = rewards + gamma * critic.predict(next_states)
  38. critic.train_on_batch(states, target_q_values)
  39. # Update the actor model
  40. action_gradients = np.array(critic.get_gradients(states, actions))
  41. actor.train_on_batch(states, action_gradients)
  42. current_state = next_state

在本例中,state_space_size和action_space_size分别是环境中的状态和操作的数量。num_episodes是轮次数。Initial_state是环境的初始状态。Take_action (current_state, action)是一个函数,它接受当前状态和操作作为输入,并返回下一个操作。


4、A2C

A2C(Advantage Actor-Critic)是一种有策略的actor-critic算法,它使用Advantage函数来更新策略。该算法实现简单,可以处理离散和连续的动作空间。

 
 
  1. import numpy as np
  2. from keras.models import Model, Sequential
  3. from keras.layers import Dense, Input
  4. from keras.optimizers import Adam
  5. from keras.utils import to_categorical
  6. # Define the actor and critic models
  7. state_input = Input(shape=(state_space_size,))
  8. actor = Dense(32, activation='relu')(state_input)
  9. actor = Dense(32, activation='relu')(actor)
  10. actor = Dense(action_space_size, activation='softmax')(actor)
  11. actor_model = Model(inputs=state_input, outputs=actor)
  12. actor_model.compile(loss='categorical_crossentropy', optimizer=Adam(lr=0.001))
  13. state_input = Input(shape=(state_space_size,))
  14. critic = Dense(32, activation='relu')(state_input)
  15. critic = Dense(32, activation='relu')(critic)
  16. critic = Dense(1, activation='linear')(critic)
  17. critic_model = Model(inputs=state_input, outputs=critic)
  18. critic_model.compile(loss='mse', optimizer=Adam(lr=0.001))
  19. for episode in range(num_episodes):
  20. current_state = initial_state
  21. done = False
  22. while not done:
  23. # Select an action using the actor model and add exploration noise
  24. action_probs = actor_model.predict(np.array([current_state]))[0]
  25. action = np.random.choice(range(action_space_size), p=action_probs)
  26. # Take the action and observe the next state and reward
  27. next_state, reward, done = take_action(current_state, action)
  28. # Calculate the advantage
  29. target_value = critic_model.predict(np.array([next_state]))[0][0]
  30. advantage = reward + gamma * target_value - critic_model.predict(np.array([current_state]))[0][0]
  31. # Update the actor model
  32. action_one_hot = to_categorical(action, action_space_size)
  33. actor_model.train_on_batch(np.array([current_state]), advantage * action_one_hot)
  34. # Update the critic model
  35. critic_model.train_on_batch(np.array([current_state]), reward + gamma * target_value)
  36. current_state = next_state

在这个例子中,actor模型是一个神经网络,它有2个隐藏层,每个隐藏层有32个神经元,具有relu激活函数,输出层具有softmax激活函数。critic模型也是一个神经网络,它有2个隐含层,每层32个神经元,具有relu激活函数,输出层具有线性激活函数。

使用分类交叉熵损失函数训练actor模型,使用均方误差损失函数训练critic模型。动作是根据actor模型预测选择的,并添加了用于探索的噪声。


5、PPO

PPO(Proximal Policy Optimization)是一种策略算法,它使用信任域优化的方法来更新策略。它在具有高维观察和连续动作空间的环境中特别有用。PPO 以其稳定性和高样品效率而著称。

 
 
  1. import numpy as np
  2. from keras.models import Model, Sequential
  3. from keras.layers import Dense, Input
  4. from keras.optimizers import Adam
  5. # Define the policy model
  6. state_input = Input(shape=(state_space_size,))
  7. policy = Dense(32, activation='relu')(state_input)
  8. policy = Dense(32, activation='relu')(policy)
  9. policy = Dense(action_space_size, activation='softmax')(policy)
  10. policy_model = Model(inputs=state_input, outputs=policy)
  11. # Define the value model
  12. value_model = Model(inputs=state_input, outputs=Dense(1, activation='linear')(policy))
  13. # Define the optimizer
  14. optimizer = Adam(lr=0.001)
  15. for episode in range(num_episodes):
  16. current_state = initial_state
  17. while not done:
  18. # Select an action using the policy model
  19. action_probs = policy_model.predict(np.array([current_state]))[0]
  20. action = np.random.choice(range(action_space_size), p=action_probs)
  21. # Take the action and observe the next state and reward
  22. next_state, reward, done = take_action(current_state, action)
  23. # Calculate the advantage
  24. target_value = value_model.predict(np.array([next_state]))[0][0]
  25. advantage = reward + gamma * target_value - value_model.predict(np.array([current_state]))[0][0]
  26. # Calculate the old and new policy probabilities
  27. old_policy_prob = action_probs[action]
  28. new_policy_prob = policy_model.predict(np.array([next_state]))[0][action]
  29. # Calculate the ratio and the surrogate loss
  30. ratio = new_policy_prob / old_policy_prob
  31. surrogate_loss = np.minimum(ratio * advantage, np.clip(ratio, 1 - epsilon, 1 + epsilon) * advantage)
  32. # Update the policy and value models
  33. policy_model.trainable_weights = value_model.trainable_weights
  34. policy_model.compile(optimizer=optimizer, loss=-surrogate_loss)
  35. policy_model.train_on_batch(np.array([current_state]), np.array([action_one_hot]))
  36. value_model.train_on_batch(np.array([current_state]), reward + gamma * target_value)
  37. current_state = next_state

6、DQN

DQN(深度 Q 网络)是一种无模型、非策略算法,它使用神经网络来逼近 Q 函数。DQN 特别适用于 Atari 游戏和其他类似问题,其中状态空间是高维的,并使用神经网络近似 Q 函数。

  1. import numpy as np
  2. from keras.models import Sequential
  3. from keras.layers import Dense, Input
  4. from keras.optimizers import Adam
  5. from collections import deque
  6. # Define the Q-network model
  7. model = Sequential()
  8. model.add(Dense(32, input_dim=state_space_size, activation='relu'))
  9. model.add(Dense(32, activation='relu'))
  10. model.add(Dense(action_space_size, activation='linear'))
  11. model.compile(loss='mse', optimizer=Adam(lr=0.001))
  12. # Define the replay buffer
  13. replay_buffer = deque(maxlen=replay_buffer_size)
  14. for episode in range(num_episodes):
  15. current_state = initial_state
  16. while not done:
  17. # Select an action using an epsilon-greedy policy
  18. if np.random.rand() < epsilon:
  19. action = np.random.randint(0, action_space_size)
  20. else:
  21. action = np.argmax(model.predict(np.array([current_state]))[0])
  22. # Take the action and observe the next state and reward
  23. next_state, reward, done = take_action(current_state, action)
  24. # Add the experience to the replay buffer
  25. replay_buffer.append((current_state, action, reward, next_state, done))
  26. # Sample a batch of experiences from the replay buffer
  27. batch = random.sample(replay_buffer, batch_size)
  28. # Prepare the inputs and targets for the Q-network
  29. inputs = np.array([x[0] for x in batch])
  30. targets = model.predict(inputs)
  31. for i, (state, action, reward, next_state, done) in enumerate(batch):
  32. if done:
  33. targets[i, action] = reward
  34. else:
  35. targets[i, action] = reward + gamma * np.max(model.predict(np.array([next_state]))[0])
  36. # Update the Q-network
  37. model.train_on_batch(inputs, targets)
  38. current_state = next_state

上面的代码,Q-network有2个隐藏层,每个隐藏层有32个神经元,使用relu激活函数。该网络使用均方误差损失函数和Adam优化器进行训练。


7、TRPO

TRPO (Trust Region Policy Optimization)是一种无模型的策略算法,它使用信任域优化方法来更新策略。它在具有高维观察和连续动作空间的环境中特别有用。

TRPO 是一个复杂的算法,需要多个步骤和组件来实现。TRPO不是用几行代码就能实现的简单算法。

所以我们这里使用实现了TRPO的现有库,例如OpenAI Baselines,它提供了包括TRPO在内的各种预先实现的强化学习算法,。

要在OpenAI Baselines中使用TRPO,我们需要安装:

 
 
pip install baselines

然后可以使用baselines库中的trpo_mpi模块在你的环境中训练TRPO代理,这里有一个简单的例子:

  1. import gym
  2. from baselines.common.vec_env.dummy_vec_env import DummyVecEnv
  3. from baselines.trpo_mpi import trpo_mpi
  4. #Initialize the environment
  5. env = gym.make("CartPole-v1")
  6. env = DummyVecEnv([lambda: env])
  7. # Define the policy network
  8. policy_fn = mlp_policy
  9. #Train the TRPO model
  10. model = trpo_mpi.learn(env, policy_fn, max_iters=1000)

我们使用Gym库初始化环境。然后定义策略网络,并调用TRPO模块中的learn()函数来训练模型。

还有许多其他库也提供了TRPO的实现,例如TensorFlow、PyTorch和RLLib。下面时一个使用TF 2.0实现的样例:

  1. import tensorflow as tf
  2. import gym
  3. # Define the policy network
  4. class PolicyNetwork(tf.keras.Model):
  5. def __init__(self):
  6. super(PolicyNetwork, self).__init__()
  7. self.dense1 = tf.keras.layers.Dense(16, activation='relu')
  8. self.dense2 = tf.keras.layers.Dense(16, activation='relu')
  9. self.dense3 = tf.keras.layers.Dense(1, activation='sigmoid')
  10. def call(self, inputs):
  11. x = self.dense1(inputs)
  12. x = self.dense2(x)
  13. x = self.dense3(x)
  14. return x
  15. # Initialize the environment
  16. env = gym.make("CartPole-v1")
  17. # Initialize the policy network
  18. policy_network = PolicyNetwork()
  19. # Define the optimizer
  20. optimizer = tf.optimizers.Adam()
  21. # Define the loss function
  22. loss_fn = tf.losses.BinaryCrossentropy()
  23. # Set the maximum number of iterations
  24. max_iters = 1000
  25. # Start the training loop
  26. for i in range(max_iters):
  27. # Sample an action from the policy network
  28. action = tf.squeeze(tf.random.categorical(policy_network(observation), 1))
  29. # Take a step in the environment
  30. observation, reward, done, _ = env.step(action)
  31. with tf.GradientTape() as tape:
  32. # Compute the loss
  33. loss = loss_fn(reward, policy_network(observation))
  34. # Compute the gradients
  35. grads = tape.gradient(loss, policy_network.trainable_variables)
  36. # Perform the update step
  37. optimizer.apply_gradients(zip(grads, policy_network.trainable_variables))
  38. if done:
  39. # Reset the environment
  40. observation = env.reset()

在这个例子中,我们首先使用TensorFlow的Keras API定义一个策略网络。然后使用Gym库和策略网络初始化环境。然后定义用于训练策略网络的优化器和损失函数。

在训练循环中,从策略网络中采样一个动作,在环境中前进一步,然后使用TensorFlow的GradientTape计算损失和梯度。然后我们使用优化器执行更新步骤。

这是一个简单的例子,只展示了如何在TensorFlow 2.0中实现TRPO。TRPO是一个非常复杂的算法,这个例子没有涵盖所有的细节,但它是试验TRPO的一个很好的起点。


总结

以上就是我们总结的7个常用的强化学习算法,这些算法并不相互排斥,通常与其他技术(如值函数逼近、基于模型的方法和集成方法)结合使用,可以获得更好的结果。

END

欢迎加入Imagination GPU与人工智能交流2群

6587717d09631b7f1ed6ded12f687ea8.jpeg

入群请加小编微信:eetrend89

(添加请备注公司名和职称)

推荐阅读

对话Imagination中国区董事长:以GPU为支点加强软硬件协同,助力数字化转型

手机芯片这个功能,有望改变市场格局!

9cc3ddd1b018e20379b18149746f5f6b.jpeg

Imagination Technologies 是一家总部位于英国的公司,致力于研发芯片和软件知识产权(IP),基于Imagination IP的产品已在全球数十亿人的电话、汽车、家庭和工作 场所中使用。获取更多物联网、智能穿戴、通信、汽车电子、图形图像开发等前沿技术信息,欢迎关注 Imagination Tech!

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/盐析白兔/article/detail/553995
推荐阅读
相关标签
  

闽ICP备14008679号