【Hierarchical RL】离线策略修正分层强化学习（HIRO）算法

举报
不去幼儿园 发表于 2024/12/05 20:45:05 2024/12/05
【摘要】 离线策略修正分层强化学习，Hierarchical Reinforcement Learning with Off-Policy Correction (HIRO) 是一种基于分层强化学习的算法，旨在解决长时间跨度和稀疏奖励问题。HIRO 特别引入了离策略（off-policy）校正机制，允许高层策略利用低层策略的经验，而不会因为低层策略的更新而产生偏差。

​

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

       【强化学习】（33）---《离线策略修正分层强化学习（HIRO）算法》

离线策略修正分层强化学习（HIRO）算法

目录

1. HIRO的核心思想

(1) 分层结构

(2) 离策略校正

2. HIRO的架构与工作机制

3. HIRO的关键公式

4. HIRO的学习方法

[Python] HIPO算法实现

算法训练代码：

算法测试代码：

[Notice]  代码说明：

5. HIRO的优点与挑战

6. HIRO的应用场景

        离线策略修正分层强化学习，Hierarchical Reinforcement Learning with Off-Policy Correction (HIRO) 是一种基于分层强化学习的算法，旨在解决长时间跨度和稀疏奖励问题。HIRO 特别引入了离策略（off-policy）校正机制，允许高层策略利用低层策略的经验，而不会因为低层策略的更新而产生偏差。

        该算法是由 K. Nachum 等人在 2018 年的论文中首次提出的。这篇论文介绍了一种新的分层强化学习方法，通过引入离策略校正机制来解决高层策略与低层策略之间的协调问题，从而应对稀疏奖励和长时间跨度的任务。论文详细描述了 HIRO 的原理、算法和实验结果，并展示了该方法在长时间跨度任务中的显著性能提升。

​

        上图展示了HIRO的设计和基础训练。较低级别的策略直接与环境交互。较高级别的策略通过高级别的动作或目标来指示较低级别的策略， gt ∈ R 它每c步重新采样一次。在中间步骤中，一个固定的目标转移函数 h决定下一步的目标。目标只是指示较低级别的策略达到特定的 这允许较低级别的策略容易地从先前的非策略经验中学习。 

1. HIRO的核心思想

        HIRO 的核心思想是在分层结构中，通过高层策略设定子目标，而低层策略则根据这些子目标执行具体的动作。为了高效利用经验和减少样本复杂度，HIRO 使用离策略校正，使得高层策略可以直接从低层策略的经验中获益。

(1) 分层结构

  • 高层策略（High-Level Policy）：每隔固定时间步选择一个子目标，指导低层策略在接下来的时间步中优化动作以实现这个目标。
  • 低层策略（Low-Level Policy）：根据高层策略设定的子目标进行动作选择，直接与环境交互，以实现该子目标。

(2) 离策略校正

        由于高层和低层策略分别进行学习，低层策略的更新会导致高层策略设定的目标无法直接应用。HIRO 通过引入校正机制，将低层策略的实际经验与高层策略的目标对齐，确保高层策略能够正确利用离策略的经验。

2. HIRO的架构与工作机制

        HIRO 的架构分为高层和低层策略，并通过校正机制确保策略的一致性。高层策略通常在更高维的隐空间中操作，而低层策略则在原始状态空间内执行动作。

(1) 高层策略的子目标生成

        高层策略基于当前状态和隐空间生成子目标( g_t )，并通过目标更新机制，将长期任务分解为多个子目标。每隔(k) 个时间步，高层策略选择新的子目标。

(2) 低层策略的动作选择

        低层策略在接收到子目标后，在每个时间步选择具体的动作( a_t )，以优化实现该子目标。低层策略直接与环境交互，并获得即时奖励。

(3) 离策略校正机制

        当低层策略执行的动作与高层策略设定的目标不一致时，HIRO 使用一个离策略校正方法来更新高层策略，使得高层策略可以利用低层策略的经验来更好地优化子目标。

3. HIRO的关键公式

(1) 高层策略的目标更新公式

        高层策略的目标更新通过学习子目标( g ) 的 Q 值函数：

[ Q_H(s, g) = r(s, g) + \gamma \cdot \mathbb{E}{s' \sim \pi{low}} \left[ Q_H(s', g') \right] ]

其中：

  • ( s )是当前状态，( g ) 是子目标。
  • ( r(s, g) )是基于高层策略的即时奖励。
  • ( \pi_{low} ) 是低层策略，根据子目标优化动作选择。

(2) 低层策略的动作选择与更新

        低层策略根据高层策略设定的子目标( g )，优化其动作。其更新公式为：

[ Q_L(s, a | g) = r(s, a) + \gamma \cdot \mathbb{E}{a' \sim \pi{low}} \left[ Q_L(s', a' | g) \right] ]

其中：

  • ( Q_L )是低层策略的 Q 值函数。
  • ( a )是低层策略选择的具体动作。

(3) 离策略校正机制公式

        校正机制使用了目标状态校正公式，将实际的低层经验( (s_t, a_t, s_{t+1}) )转换为高层策略的目标状态 ( g_t )：

[ g_t = s_{t+k} - s_t ]

其中，( s_{t+k} )是未来( k ) 个时间步后的状态，用于校正高层策略设定的目标。

4. HIRO的学习方法

        HIRO 通过离策略学习和动态校正的结合，使高层和低层策略能够互相独立优化，而不产生偏差。学习方法如下：

  1. 离策略数据收集：低层策略与环境交互，收集多样化的数据集。高层策略可以直接利用这些数据集进行优化。
  2. 离策略校正：通过校正机制，调整高层策略设定的目标，使得高层策略能够正确利用低层策略的经验。
  3. 策略优化：高层策略根据校正后的数据更新子目标选择策略；低层策略则使用子目标优化其动作选择策略。

[Python] HIPO算法实现

        HIRO 是一种增强型分层强化学习算法，使用两层策略结构：

  • 高层策略（Manager）：根据环境状态生成子目标。
  • 低层策略（Worker）：根据当前状态和高层策略的子目标选择具体动作，并完成子任务。

        HIRO 通过使用 Off-Policy Correction 方法解决分层策略中的动作不匹配问题，使得不同层次的策略能够在异步更新时更加高效

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

算法训练代码：

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


# 高层策略网络（生成子目标）
class HighLevelPolicy(nn.Module):
    def __init__(self, state_dim, goal_dim):
        super(HighLevelPolicy, self).__init__()
        self.fc = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, goal_dim)
        )

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

# 低层策略网络（根据子目标生成动作）
class LowLevelPolicy(nn.Module):
    def __init__(self, state_dim, goal_dim, action_dim):
        super(LowLevelPolicy, self).__init__()
        self.fc = nn.Sequential(
            nn.Linear(state_dim + goal_dim, 128),
            nn.ReLU(),
            nn.Linear(128, action_dim)
        )

    def forward(self, state, goal):
        x = torch.cat([state, goal], dim=-1)
        return self.fc(x)

# 低层策略类
class Worker:
    def __init__(self, policy_net):
        self.policy_net = policy_net

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

# HIRO 代理类
class HIROAgent:
    def __init__(self, state_dim, goal_dim, action_dim):
        self.high_level_policy = HighLevelPolicy(state_dim, goal_dim)
        self.low_level_policy = LowLevelPolicy(state_dim, goal_dim, action_dim)
        self.high_level_optimizer = optim.Adam(self.high_level_policy.parameters(), lr=0.001)
        self.low_level_optimizer = optim.Adam(self.low_level_policy.parameters(), lr=0.001)
        self.gamma = 0.99
        self.worker = Worker(self.low_level_policy)
        self.goal_dim = goal_dim

    def select_goal(self, state):
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        goal = self.high_level_policy(state_tensor).detach().numpy().squeeze()
        return goal

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

        current_goal = self.high_level_policy(state_tensor)
        next_goal = self.high_level_policy(next_state_tensor).detach()

        target = reward + (1 - done) * self.gamma * next_goal
        loss = nn.functional.mse_loss(current_goal, target)

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

    def train_low_level_policy(self, state, goal, action, reward, next_state, done):
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        goal_tensor = torch.FloatTensor(goal).unsqueeze(0)
        next_state_tensor = torch.FloatTensor(next_state).unsqueeze(0)

        q_values = self.low_level_policy(state_tensor, goal_tensor)
        next_q_values = self.low_level_policy(next_state_tensor, goal_tensor)

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

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

# 训练函数
def train_hiro(env, agent, num_episodes=1000):
    for episode in range(num_episodes):
        state, _ = env.reset()
        total_reward = 0
        for t in range(200):  # 每个 episode 最多 200 步
            goal = agent.select_goal(state)  # 高层生成子目标
            action = agent.worker.select_action(state, goal)  # 低层根据子目标生成动作

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

            # 更新高层策略
            agent.train_high_level_policy(state, reward, next_state, done)
            # 更新低层策略
            agent.train_low_level_policy(state, goal, action, 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

# 实例化 HIRO 代理并训练
goal_dim = 2  # 选择一个简单的2维目标空间
agent = HIROAgent(state_dim, goal_dim, action_dim)

# 训练代理
train_hiro(env, agent, num_episodes=1000)

算法测试代码：

# 测试函数
def test_hiro(env, agent, num_episodes=5):
    for episode in range(num_episodes):
        state, _ = env.reset()
        total_reward = 0
        for t in range(200):
            env.render()  # 显示动画

            goal = agent.select_goal(state)
            action = agent.worker.select_action(state, goal)

            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_hiro(env, agent, num_episodes=5)

[Notice]  代码说明：

  1. HighLevelPolicy：高层策略网络生成子目标。
  2. LowLevelPolicy：低层策略网络根据子目标和当前状态生成动作。
  3. Worker：低层策略执行者类，用于选择动作。
  4. HIROAgent：代理类，包含高层和低层策略的训练方法。
  5. train_hiro：训练函数，通过高层策略生成子目标，低层策略生成动作。
  6. test_hiro：测试函数，在 CartPole 环境中显示动画。

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

5. HIRO的优点与挑战

(1) 优点：

  • 样本高效性：离策略校正机制使得 HIRO 可以更高效地利用低层策略的经验，提高样本利用率。
  • 适应长时间跨度任务：通过动态校正和多层策略分解，HIRO 能够更好地应对长时间跨度的复杂任务。
  • 鲁棒性：HIRO 通过校正机制有效避免了传统 HRL 在离策略学习中的偏差问题，提高了策略的鲁棒性。

(2) 挑战：

  • 校正机制复杂度：校正机制的实现可能较为复杂，尤其在高维状态空间中。
  • 子目标定义的有效性：校正机制依赖子目标的正确性，因此合理定义子目标至关重要。

6. HIRO的应用场景

        HIRO 适用于稀疏奖励、长时间跨度、以及需要动态调整子任务的复杂任务，如：

  • 复杂机器人控制：HIRO 可以处理长时间跨度的控制任务，例如多关节机器人操作、长距离导航等。
  • 高维策略游戏：在复杂游戏场景中，HIRO 可以通过分解和校正机制应对长期策略问题。
  • 多步骤规划与决策任务：例如资源分配和路径规划任务，HIRO 通过离策略校正优化决策效率。

        HIRO 是一种灵活且高效的分层强化学习方法，在多个领域中取得了显著的性能提升。

参考文献：K. Nachum, S. Gu, H. Lee, S. Levine (2018). Data-Efficient 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+优质创作者共同成长
立即加入