【Hierarchical RL】动态分层强化学习（DHRL）算法

举报
不去幼儿园 发表于 2024/12/05 20:43:23 2024/12/05
【摘要】 动态分层强化学习，Dynamic Hierarchical Reinforcement Learning (DHRL) 是一种自适应分层强化学习算法，其目标是根据任务和环境的复杂性动态地构建、修改和利用分层策略。DHRL 不仅仅是预定义层次结构的简单执行，而是允许代理在学习过程中根据需要动态生成和调整分层策略，从而实现更好的任务分解和高效学习。 DHRL 扩展了传统的分层强化学习（HRL），通过动

​

   📢本篇文章是博主强化学习（RL）领域学习时，用于个人学习、研究或者欣赏使用，并基于博主对相关等领域的一些理解而记录的学习摘录和笔记，若有不当和侵权之处，指出后将会立即改正，还望谅解。文章分类在👉强化学习专栏：

       【强化学习】（32）---《动态分层强化学习（DHRL）算法》

动态分层强化学习（DHRL）算法

目录

1. DHRL的核心思想

(1) 动态层次生成

(2) 策略动态调整

(3) 分层的自适应性

2. DHRL的架构与工作机制

(1) 高层与低层交互

(2) 任务分解与子任务生成

3. DHRL的动态调整机制

4. DHRL的关键公式

5. DHRL的学习方法

[Python] DHRL算法实现

DHRL 的主要步骤：

实现思路：

算法训练代码：

算法测试代码：

[Notice]  代码说明

6. DHRL的优点与挑战

7. DHRL的应用场景

        动态分层强化学习，Dynamic Hierarchical Reinforcement Learning (DHRL) 是一种自适应分层强化学习算法，其目标是根据任务和环境的复杂性动态地构建、修改和利用分层策略。DHRL 不仅仅是预定义层次结构的简单执行，而是允许代理在学习过程中根据需要动态生成和调整分层策略，从而实现更好的任务分解和高效学习。

        DHRL 扩展了传统的分层强化学习（HRL），通过动态调整层次和策略，使其适应环境中的变化和不确定性。这种方法能够处理复杂任务，特别是那些需要灵活调整策略或面临多种不同子任务的情景。

​

        上图展示了DHRL与以前基于图的HRL方法的区别。DHRL算法显式地包含了两层之间的图结构，而之前的方法只使用图来训练高级策略或获取路径点。 

1. DHRL的核心思想

        DHRL 的核心思想是通过动态生成和调整层次结构来处理不同的任务和状态。与传统的 HRL 不同，DHRL 并不预定义固定的层次结构，而是通过不断地优化和调整策略来实现任务的分解和动态层次生成。这使得算法更具灵活性，能够适应变化的环境。

(1) 动态层次生成

  • 代理在学习过程中自动识别并生成新的层次。每当代理遇到一个新的复杂任务时，它可以通过生成新的子任务或选项来分解这个任务。
  • 通过对历史数据和当前环境的分析，代理可以动态生成或调整层次，以优化长期回报。

(2) 策略动态调整

  • 在 DHRL 中，策略不仅仅在学习初期进行设计，代理可以在整个学习过程中不断调整其高层和低层策略，以应对环境变化和任务复杂性。

(3) 分层的自适应性

  • DHRL 的核心是自适应层次生成。不同于传统的分层结构，DHRL 允许根据状态和环境的复杂性动态调整高层与低层的交互模式。
  • 代理根据任务需要选择是继续分解任务还是直接执行底层动作。

2. DHRL的架构与工作机制

        DHRL 的分层架构与传统的 HRL 类似，但更注重动态生成和调整。DHRL 通常有两层或多层结构：

  • 高层策略（High-Level Policy）：负责任务的全局规划和子任务生成。该策略会根据当前的状态和任务生成新的子任务或分配已经学习的子任务。
  • 低层策略（Low-Level Policy）：负责执行具体的动作，或者在任务复杂度较高时，通过动态层次生成新的子任务。

(1) 高层与低层交互

  • 高层策略设定全局目标或任务，并将其分解为子任务。
  • 低层策略执行高层设定的子任务，若任务复杂度较高，低层策略可以生成新的子任务，继续执行分解。

(2) 任务分解与子任务生成

  • DHRL 可以根据任务的复杂性生成新的子任务。高层策略根据环境和当前任务的状态，动态生成适当的子任务或子策略。
  • 子任务生成的过程基于环境反馈和历史数据，动态调整任务层次。

3. DHRL的动态调整机制

        DHRL 的动态调整机制基于以下几个核心组件：

  1. 状态评估与层次生成：当代理遇到新任务或环境状态时，它会根据历史学习经验生成新的层次。如果任务足够复杂，则通过分层结构进行任务分解；否则，直接选择底层动作。
  2. 子任务复用与优化：代理在执行子任务时，可以复用已学到的子任务策略，从而减少重复学习。这种方法允许在多任务环境中复用策略，从而提高学习效率。
  3. 层次调整与策略优化：当环境或任务发生变化时，DHRL 允许代理调整已有的层次结构，并动态调整高层或低层策略，以更好地适应当前任务。

4. DHRL的关键公式

        DHRL 的策略优化涉及到高层和低层的策略更新，这与标准的 Q-learning 相似，但引入了动态生成与终止条件的扩展。

(1) 高层策略的更新

        高层策略( \pi_H )选择子任务( o )或直接选择底层动作。其值函数更新为：

[ Q_H(s, o) = \mathbb{E}\left[ \sum_{t=0}^{T} \gamma^t r(s_t, o_t) \right] ]

其中：

  • ( s )是当前状态。
  • ( o )是高层选择的子任务。
  • ( r(s_t, o_t) ) 是子任务执行过程中的奖励。

(2) 低层策略的更新

        低层策略 ( \pi_L )负责在每个时间步内选择具体动作。当低层策略执行子任务时，Q 值更新为：

[ Q_L(s, a | o) = r + \gamma^\tau \max_{a'} Q_L(s', a' | o) ]

其中：

  • ( a )是低层执行的具体动作。
  • ( \tau )是动作执行的时间步数。

5. DHRL的学习方法

        DHRL 的学习过程结合了分层强化学习的多层策略，同时动态调整和生成策略。学习过程如下：

  1. 任务评估：当代理遇到新任务或复杂环境时，评估任务的复杂度并决定是否生成新的层次。
  2. 动态生成层次：若任务复杂，则通过 DHRL 的层次生成机制生成新的子任务或新的高层策略。
  3. 策略优化：通过高层和低层的交互，代理不断优化策略，并根据环境变化动态调整层次结构。
  4. 子任务复用：在多任务场景中，DHRL 允许代理复用已学习的子任务策略，从而减少学习时间。

[Python] DHRL算法实现

        DHRL 是一种在层次结构上进行学习的强化学习方法。它通过动态地构建和调整层级结构，从而高效地解决复杂任务。与固定的分层架构不同，DHRL 可以在训练过程中动态构建子任务和策略，适应环境的变化。

        在经典的 CartPole 环境中，我们可以设计一个简单的 DHRL 算法，将问题划分为多个子任务，并在每个子任务中使用强化学习算法来求解。每个子任务会有不同的策略，父层控制何时切换子任务，动态分配任务层次。

        🔥若是下面代码复现困难或者有问题，欢迎评论区留言；需要以整个项目形式的代码，请在评论区留下您的邮箱📌，以便于及时分享给您（私信难以及时回复）。

DHRL 的主要步骤：

  1. 顶层管理器（Controller）：负责根据当前状态选择不同的子任务（低层策略）。
  2. 低层策略（Sub-policies）：每个子任务对应一个低层策略，负责执行具体动作。
  3. 动态调整层级：根据训练过程中的表现，动态调整任务的分配和策略。

实现思路：

  • 顶层控制器管理多个子任务（例如：保持平衡、移动到左、移动到右）。
  • 子任务使用独立的策略网络来执行特定的动作。
  • 顶层控制器通过 Q-learning 或 Actor-Critic 方法选择子任务，并监控其执行情况。

算法训练代码：

"""《DHRL算法实现》
    时间：2024.10
    环境：CartPole
    作者：不去幼儿园
"""
import gym
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import random


# 高层控制器网络
class ControllerNet(nn.Module):
	def __init__(self, state_dim, num_subtasks):
		super(ControllerNet, self).__init__()
		self.fc = nn.Sequential(
			nn.Linear(state_dim, 128),
			nn.ReLU(),
			nn.Linear(128, num_subtasks)
		)

	def forward(self, state):
		return self.fc(state)


# 子任务策略网络
class SubtaskPolicyNet(nn.Module):
	def __init__(self, state_dim, action_dim):
		super(SubtaskPolicyNet, self).__init__()
		self.fc = nn.Sequential(
			nn.Linear(state_dim, 128),
			nn.ReLU(),
			nn.Linear(128, action_dim)
		)

	def forward(self, state):
		return self.fc(state)


# 低层子任务类
class Subtask:
	def __init__(self, policy_net):
		self.policy_net = policy_net

	def select_action(self, state):
		state = torch.FloatTensor(state).unsqueeze(0)
		action_probs = self.policy_net(state)
		action = torch.argmax(action_probs).item()
		return action


# 动态分层强化学习智能体
class DHRLAgent:
	def __init__(self, state_dim, action_dim, num_subtasks):
		self.controller = ControllerNet(state_dim, num_subtasks)
		self.subtasks = [Subtask(SubtaskPolicyNet(state_dim, action_dim)) for _ in range(num_subtasks)]
		self.controller_optimizer = optim.Adam(self.controller.parameters(), lr=0.001)
		self.subtask_optimizers = [optim.Adam(subtask.policy_net.parameters(), lr=0.001) for subtask in self.subtasks]
		self.gamma = 0.99

	def select_subtask(self, state):
		state_tensor = torch.FloatTensor(state).unsqueeze(0)
		subtask_probs = self.controller(state_tensor)
		selected_subtask = torch.argmax(subtask_probs).item()
		return selected_subtask

	def train_subtask(self, subtask_idx, state, action, reward, next_state, done):
		optimizer = self.subtask_optimizers[subtask_idx]
		subtask = self.subtasks[subtask_idx]
		state_tensor = torch.FloatTensor(state).unsqueeze(0)
		next_state_tensor = torch.FloatTensor(next_state).unsqueeze(0)

		# 计算当前 Q 值
		q_values = subtask.policy_net(state_tensor)
		next_q_values = subtask.policy_net(next_state_tensor)

		target_q_value = reward + (1 - done) * self.gamma * torch.max(next_q_values).item()
		loss = nn.functional.mse_loss(q_values[0, action], torch.tensor(target_q_value))

		optimizer.zero_grad()
		loss.backward()
		optimizer.step()

	def train_controller(self, state, subtask_idx, reward, next_state, done):
		state_tensor = torch.FloatTensor(state).unsqueeze(0)
		next_state_tensor = torch.FloatTensor(next_state).unsqueeze(0)

		# 计算高层控制器 Q 值
		controller_q_values = self.controller(state_tensor)
		next_controller_q_values = self.controller(next_state_tensor)

		target_q_value = reward + (1 - done) * self.gamma * torch.max(next_controller_q_values).item()
		loss = nn.functional.mse_loss(controller_q_values[0, subtask_idx], torch.tensor(target_q_value))

		self.controller_optimizer.zero_grad()
		loss.backward()
		self.controller_optimizer.step()


# 训练函数
def train_dhrl(env, agent, num_episodes=1000):
	for episode in range(num_episodes):
		state, _ = env.reset()
		total_reward = 0

		for t in range(200):  # 每个 episode 最多 200 步
			subtask_idx = agent.select_subtask(state)  # 选择当前子任务
			action = agent.subtasks[subtask_idx].select_action(state)  # 选择动作

			next_state, reward, done, _, _ = env.step(action)
			total_reward += reward

			# 训练子任务
			agent.train_subtask(subtask_idx, state, action, reward, next_state, done)
			# 训练高层控制器
			agent.train_controller(state, subtask_idx, reward, next_state, done)

			state = next_state

			if done:
				break

		print(f'Episode {episode}: Total Reward: {total_reward}')


# 环境设置
env = gym.make('CartPole-v1')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n

# 实例化 DHRL 智能体并训练
num_subtasks = 2  # 两个子任务，左右移动
agent = DHRLAgent(state_dim, action_dim, num_subtasks)

# 训练智能体
train_dhrl(env, agent, num_episodes=1000)

算法测试代码：

# 测试函数
def test_dhrl(env, agent, num_episodes=5):
	for episode in range(num_episodes):
		state, _ = env.reset()
		total_reward = 0

		for t in range(200):
			env.render()  # 显示动画

			subtask_idx = agent.select_subtask(state)
			action = agent.subtasks[subtask_idx].select_action(state)

			next_state, reward, done, _, _ = env.step(action)
			total_reward += reward

			state = next_state

			if done:
				break

		print(f'Test Episode {episode}: Total Reward: {total_reward}')

	env.close()


# 测试代理并显示动画
env = gym.make('CartPole-v1', render_mode="human")
test_dhrl(env, agent, num_episodes=5)

[Notice]  代码说明

  1. ControllerNet：高层管理器，用于根据当前状态选择子任务。
  2. SubtaskPolicyNet：每个子任务对应的策略网络，选择具体动作。
  3. Subtask：封装子任务的策略网络，并定义 select_action 方法选择动作。
  4. DHRLAgent：DHRL代理类，封装了高层管理器和子任务。该类实现了子任务和管理器的训练方法。
  5. train_dhrl：训练过程，其中每一步根据当前状态选择子任务，执行动作，并更新管理器和子任务的策略。
  6. test_dhrl：测试过程，运行一段时间，并显示 CartPole 的动画。

运行说明：

  • 训练过程：代理在 train_dhrl 中进行训练，并调整管理器和子任务的策略。
  • 测试过程：在 test_dhrl 中显示动画，观察 DHRL 代理在 CartPole 环境中的表现。

        由于博文主要为了介绍相关算法的原理和应用的方法，缺乏对于实际效果的关注，算法可能在上述环境中的效果不佳，一是算法不适配上述环境，二是算法未调参和优化，三是等等。上述代码用于了解和学习算法足够了，但若是想直接将上面代码应用于实际项目中，还需要进行修改。

6. DHRL的优点与挑战

(1) 优点：

  • 动态适应任务复杂性：DHRL 通过动态生成和调整分层结构，能够处理不同复杂度的任务，并根据需要生成新的子任务。
  • 复用策略：代理能够复用已学习的子任务和策略，从而提高多任务学习的效率。
  • 应对环境变化：DHRL 能够在不断变化的环境中动态调整策略，使其更加灵活和高效。

(2) 挑战：

  • 层次生成的复杂性：生成合适的分层结构是一个复杂的问题，如何平衡任务分解与直接动作选择是一个关键挑战。
  • 子任务的自动化发现：如何自动生成有效的子任务和策略，尤其是在复杂任务中，仍然是一个未完全解决的问题。

7. DHRL的应用场景

DHRL 适用于复杂环境中的长时间规划任务，特别是需要灵活应对环境变化和多任务复用的场景：

  • 机器人控制：DHRL 能够根据任务复杂性动态生成子任务，适合复杂的机器人控制任务，例如动态导航和操作控制。
  • 策略游戏：在需要长时间规划的游戏中，DHRL 通过动态生成策略来应对不同的游戏场景。

参考文献：

DHRL: A Graph-Based Approach for Long-Horizon and Sparse Hierarchical Reinforcement Learning 

  🔥想了解更多分层强化学习的文章，请查看文章：

【RL Latest Tech】分层强化学习（Hierarchical RL）

     文章若有不当和不正确之处，还望理解与指出。由于部分文字、图片等来源于互联网，无法核实真实出处，如涉及相关争议，请联系博主删除。如有错误、疑问和侵权，欢迎评论留言联系作者，或者添加VX：Rainbook_2，联系作者。✨

​

【版权声明】本文为华为云社区用户原创内容，未经允许不得转载，如需转载请自行联系原作者进行授权。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
Python 人工智能 机器学习 神经网络 网络
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入