0
0

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

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

1、Q-learning

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

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

  1.  importnumpyasnp
  2.  
  3.  # Define the Q-table and the learning rate
  4.  Q=np.zeros((state_space_size, action_space_size))
  5.  alpha=0.1
  6.  
  7.  # Define the exploration rate and discount factor
  8.  epsilon=0.1
  9.  gamma=0.99
  10.  
  11.  forepisodeinrange(num_episodes):
  12.      current_state=initial_state
  13.      whilenotdone:
  14.          # Choose an action using an epsilon-greedy policy
  15.          ifnp.random.uniform(0, 1) <epsilon:
  16.              action=np.random.randint(0, action_space_size)
  17.          else:
  18.              action=np.argmax(Q[current_state])
  19.  
  20.          # Take the action and observe the next state and reward
  21.          next_state, reward, done=take_action(current_state, action)
  22.  
  23.          # Update the Q-table using the Bellman equation
  24.          Q[current_state, action] =Q[current_state, action] +alpha* (reward+gamma*np.max(Q[next_state]) -Q[current_state, action])
  25.  
  26.          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.  importnumpyasnp
  2.  
  3.  # Define the Q-table and the learning rate
  4.  Q=np.zeros((state_space_size, action_space_size))
  5.  alpha=0.1
  6.  
  7.  # Define the exploration rate and discount factor
  8.  epsilon=0.1
  9.  gamma=0.99
  10.  
  11.  forepisodeinrange(num_episodes):
  12.      current_state=initial_state
  13.      action=epsilon_greedy_policy(epsilon, Q, current_state)
  14.      whilenotdone:
  15.          # Take the action and observe the next state and reward
  16.          next_state, reward, done=take_action(current_state, action)
  17.          # Choose next action using epsilon-greedy policy
  18.          next_action=epsilon_greedy_policy(epsilon, Q, next_state)
  19.          # Update the Q-table using the Bellman equation
  20.          Q[current_state, action] =Q[current_state, action] +alpha* (reward+gamma*Q[next_state, next_action] -Q[current_state, action])
  21.          current_state=next_state
  22.          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.  importnumpyasnp
  2.  fromkeras.modelsimportModel, Sequential
  3.  fromkeras.layersimportDense, Input
  4.  fromkeras.optimizersimportAdam
  5.  
  6.  # Define the actor and critic models
  7.  actor=Sequential()
  8.  actor.add(Dense(32, input_dim=state_space_size, activation='relu'))
  9.  actor.add(Dense(32, activation='relu'))
  10.  actor.add(Dense(action_space_size, activation='tanh'))
  11.  actor.compile(loss='mse', optimizer=Adam(lr=0.001))
  12.  
  13.  critic=Sequential()
  14.  critic.add(Dense(32, input_dim=state_space_size, activation='relu'))
  15.  critic.add(Dense(32, activation='relu'))
  16.  critic.add(Dense(1, activation='linear'))
  17.  critic.compile(loss='mse', optimizer=Adam(lr=0.001))
  18.  
  19.  # Define the replay buffer
  20.  replay_buffer= []
  21.  
  22.  # Define the exploration noise
  23.  exploration_noise=OrnsteinUhlenbeckProcess(size=action_space_size, theta=0.15, mu=0, sigma=0.2)
  24.  
  25.  forepisodeinrange(num_episodes):
  26.      current_state=initial_state
  27.      whilenotdone:
  28.          # Select an action using the actor model and add exploration noise
  29.          action=actor.predict(current_state)[0] +exploration_noise.sample()
  30.          action=np.clip(action, -1, 1)
  31.  
  32.          # Take the action and observe the next state and reward
  33.          next_state, reward, done=take_action(current_state, action)
  34.  
  35.          # Add the experience to the replay buffer
  36.          replay_buffer.append((current_state, action, reward, next_state, done))
  37.  
  38.          # Sample a batch of experiences from the replay buffer
  39.          batch=sample(replay_buffer, batch_size)
  40.  
  41.          # Update the critic model
  42.          states=np.array([x[0] forxinbatch])
  43.          actions=np.array([x[1] forxinbatch])
  44.          rewards=np.array([x[2] forxinbatch])
  45.          next_states=np.array([x[3] forxinbatch])
  46.  
  47.          target_q_values=rewards+gamma*critic.predict(next_states)
  48.          critic.train_on_batch(states, target_q_values)
  49.  
  50.          # Update the actor model
  51.          action_gradients=np.array(critic.get_gradients(states, actions))
  52.          actor.train_on_batch(states, action_gradients)
  53.  
  54.          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.  importnumpyasnp
  2.  fromkeras.modelsimportModel, Sequential
  3.  fromkeras.layersimportDense, Input
  4.  fromkeras.optimizersimportAdam
  5.  fromkeras.utilsimportto_categorical
  6.  
  7.  # Define the actor and critic models
  8.  state_input=Input(shape=(state_space_size,))
  9.  actor=Dense(32, activation='relu')(state_input)
  10.  actor=Dense(32, activation='relu')(actor)
  11.  actor=Dense(action_space_size, activation='softmax')(actor)
  12.  actor_model=Model(inputs=state_input, outputs=actor)
  13.  actor_model.compile(loss='categorical_crossentropy', optimizer=Adam(lr=0.001))
  14.  
  15.  state_input=Input(shape=(state_space_size,))
  16.  critic=Dense(32, activation='relu')(state_input)
  17.  critic=Dense(32, activation='relu')(critic)
  18.  critic=Dense(1, activation='linear')(critic)
  19.  critic_model=Model(inputs=state_input, outputs=critic)
  20.  critic_model.compile(loss='mse', optimizer=Adam(lr=0.001))
  21.  
  22.  forepisodeinrange(num_episodes):
  23.      current_state=initial_state
  24.      done=False
  25.      whilenotdone:
  26.          # Select an action using the actor model and add exploration noise
  27.          action_probs=actor_model.predict(np.array([current_state]))[0]
  28.          action=np.random.choice(range(action_space_size), p=action_probs)
  29.  
  30.          # Take the action and observe the next state and reward
  31.          next_state, reward, done=take_action(current_state, action)
  32.  
  33.          # Calculate the advantage
  34.          target_value=critic_model.predict(np.array([next_state]))[0][0]
  35.          advantage=reward+gamma*target_value-critic_model.predict(np.array([current_state]))[0][0]
  36.  
  37.          # Update the actor model
  38.          action_one_hot=to_categorical(action, action_space_size)
  39.          actor_model.train_on_batch(np.array([current_state]), advantage*action_one_hot)
  40.  
  41.          # Update the critic model
  42.          critic_model.train_on_batch(np.array([current_state]), reward+gamma*target_value)
  43.  
  44.          current_state=next_state

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

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

5、PPO

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

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

6、DQN

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

  1.  importnumpyasnp
  2.  fromkeras.modelsimportSequential
  3.  fromkeras.layersimportDense, Input
  4.  fromkeras.optimizersimportAdam
  5.  fromcollectionsimportdeque
  6.  
  7.  # Define the Q-network model
  8.  model=Sequential()
  9.  model.add(Dense(32, input_dim=state_space_size, activation='relu'))
  10.  model.add(Dense(32, activation='relu'))
  11.  model.add(Dense(action_space_size, activation='linear'))
  12.  model.compile(loss='mse', optimizer=Adam(lr=0.001))
  13.  
  14.  # Define the replay buffer
  15.  replay_buffer=deque(maxlen=replay_buffer_size)
  16.  
  17.  forepisodeinrange(num_episodes):
  18.      current_state=initial_state
  19.      whilenotdone:
  20.          # Select an action using an epsilon-greedy policy
  21.          ifnp.random.rand() <epsilon:
  22.              action=np.random.randint(0, action_space_size)
  23.          else:
  24.              action=np.argmax(model.predict(np.array([current_state]))[0])
  25.  
  26.          # Take the action and observe the next state and reward
  27.          next_state, reward, done=take_action(current_state, action)
  28.  
  29.          # Add the experience to the replay buffer
  30.          replay_buffer.append((current_state, action, reward, next_state, done))
  31.  
  32.          # Sample a batch of experiences from the replay buffer
  33.          batch=random.sample(replay_buffer, batch_size)
  34.  
  35.          # Prepare the inputs and targets for the Q-network
  36.          inputs=np.array([x[0] forxinbatch])
  37.          targets=model.predict(inputs)
  38.          fori, (state, action, reward, next_state, done) inenumerate(batch):
  39.              ifdone:
  40.                  targets[i, action] =reward
  41.              else:
  42.                  targets[i, action] =reward+gamma*np.max(model.predict(np.array([next_state]))[0])
  43.  
  44.          # Update the Q-network
  45.          model.train_on_batch(inputs, targets)
  46.  
  47.          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，我们需要安装:

  1.  pip install baselines

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

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

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

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

  1.  importtensorflowastf
  2.  importgym
  3.  
  4.  # Define the policy network
  5.  classPolicyNetwork(tf.keras.Model):
  6.      def__init__(self):
  7.          super(PolicyNetwork, self).__init__()
  8.          self.dense1=tf.keras.layers.Dense(16, activation='relu')
  9.          self.dense2=tf.keras.layers.Dense(16, activation='relu')
  10.          self.dense3=tf.keras.layers.Dense(1, activation='sigmoid')
  11.  
  12.      defcall(self, inputs):
  13.          x=self.dense1(inputs)
  14.          x=self.dense2(x)
  15.          x=self.dense3(x)
  16.          returnx
  17.  
  18.  # Initialize the environment
  19.  env=gym.make("CartPole-v1")
  20.  
  21.  # Initialize the policy network
  22.  policy_network=PolicyNetwork()
  23.  
  24.  # Define the optimizer
  25.  optimizer=tf.optimizers.Adam()
  26.  
  27.  # Define the loss function
  28.  loss_fn=tf.losses.BinaryCrossentropy()
  29.  
  30.  # Set the maximum number of iterations
  31.  max_iters=1000
  32.  
  33.  # Start the training loop
  34.  foriinrange(max_iters):
  35.      # Sample an action from the policy network
  36.      action=tf.squeeze(tf.random.categorical(policy_network(observation), 1))
  37.  
  38.      # Take a step in the environment
  39.      observation, reward, done, _=env.step(action)
  40.  
  41.      withtf.GradientTape() astape:
  42.          # Compute the loss
  43.          loss=loss_fn(reward, policy_network(observation))
  44.  
  45.      # Compute the gradients
  46.      grads=tape.gradient(loss, policy_network.trainable_variables)
  47.  
  48.      # Perform the update step
  49.      optimizer.apply_gradients(zip(grads, policy_network.trainable_variables))
  50.  
  51.      ifdone:
  52.          # Reset the environment
  53.          observation=env.reset()

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

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

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

总结

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

作者:Siddhartha Pramanik

标签： 机器学习 强化学习 深度学习
登录后发布评论

“7个流行的强化学习算法及代码实现”的评论:

还没有评论
关于作者
...
Deephub
更多文章请关注公众号：Deephub-IMBA
相关阅读

2025年GitHub平台上的十大开源MCP服务器汇总分析

Crawl4AI：为大语言模型打造的开源网页数据采集工具

Triton入门教程：安装与编写和运行简单Triton内核

用离散标记重塑人体姿态：VQ-VAE实现关键点组合关系编码

MCP与A2A协议比较：人工智能系统互联与协作的技术基础架构

比扩散策略更高效的生成模型：流匹配的理论基础与Pytorch代码实现

从零实现基于扩散模型的文本到视频生成系统：技术详解与Pytorch代码实现