【RL Latest Tech】分层强化学习：MAXQ分解算法

MAXQ分解是一种用于分层强化学习（Hierarchical Reinforcement Learning, HRL）的算法，由Thomas G. Dietterich提出。该算法通过将复杂的任务分解成更小的子任务来简化问题，并利用这些子任务来构建更复杂的策略。主要思想是将一个复杂的Markov决策过程（MDP）分解成一系列嵌套的子MDP，以便更容易解决。MAXQ算法引入了一种分层的结构，将原始任

不去幼儿园

1353人浏览 · 2024-10-03 11:29:08

不去幼儿园 · 2024-10-03 11:29:08 发布

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

【强化学习】（23）---《分层强化学习：MAXQ分解算法》

分层强化学习：MAXQ分解算法

1. MAXQ分解算法的核心思想

2. 任务层次结构（Task Hierarchy）

MAXQ分解是一种用于分层强化学习（Hierarchical Reinforcement Learning, HRL）的算法，由Thomas G. Dietterich提出。该算法通过将复杂的任务分解成更小的子任务来简化问题，并利用这些子任务来构建更复杂的策略。

1. MAXQ分解算法的核心思想

MAXQ分解的主要思想是将一个复杂的Markov决策过程（MDP）分解成一系列嵌套的子MDP，以便更容易解决。MAXQ算法引入了一种分层的结构，将原始任务逐步分解为多个子任务，从而形成一个任务树（task hierarchy），并通过各个子任务的求解来最终解决整个任务。

2. 任务层次结构（Task Hierarchy）

在MAXQ分解算法中，任务被组织成一个分层结构，其中每一个节点都是一个子任务。任务层次结构的关键特征包括：

根任务（Root Task）：这是整个任务的顶层，即原始的MDP问题。
子任务（Subtasks）：根任务可以分解成一系列子任务，每个子任务又可以进一步分解成更小的子任务，直到最底层为不可再分的原子动作（Primitive Actions）。
子任务的目标：每个子任务都有一个特定的目标（Goal），即在某个状态下完成该子任务。

任务树中的每个子任务对应一个子MDP，包含其状态、动作和奖励结构。子任务的动作可以是原子动作（如“向前移动一步”），也可以是其他子任务（递归地分解）。

3. MAXQ值函数分解

MAXQ算法的核心是将值函数（Value Function）分解成一系列子任务的值函数。这种分解被称为“MAXQ值函数分解”，包括以下两个部分：

Completion Function $( C(s, a) )$ ：表示在给定状态 s 下，执行子任务 a 的完成过程中所累积的期望奖励。
Q函数 $( V(s) )$ ：描述了在给定状态 s 下，不同子任务 a 的选择带来的期望回报。

公式表示为：

$[ Q(s, a) = C(s, a) + V(s) ]$

其中， $( Q(s, a) )$ 表示在状态 s 选择动作（子任务） a 的值。

4. MAXQ学习过程

MAXQ分解的学习过程是通过策略梯度或Q-learning等强化学习算法来进行的。学习过程包括以下几个步骤：

策略学习：对于每个子任务，学习其最优策略，使得完成该子任务的期望奖励最大化。
值函数更新：通过子任务的执行和奖励反馈来更新对应的值函数 $( Q(s, a) )$ 和完成函数 $( C(s, a) )$ 。
分层执行：在执行过程中，首先选择顶层任务，然后递归地选择各个子任务，直到执行到原子动作。

[Python] MAXQ分解算法实现

在 CartPole 环境中使用 MAXQ 分解算法的 Python 实现。MAXQ 分解是分层强化学习的一种方法，它通过将任务分解为多个子任务，从而减少问题的复杂性，达到更高效的学习效果。具体来说，MAXQ 会将任务分解成多个层次的子任务，智能体会根据不同的子任务学习不同的策略。

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

主要步骤：

设置 CartPole 环境。
定义 MAXQ 分解结构，将任务分解为子任务。
在各个子任务中分别学习Q值。
通过递归结构自上而下地选择动作，并利用低层策略解决高层问题。
通过Q值更新来优化分层结构中的各个任务。

训练代码实现：

"""《MAXQ分解算法实现》
    时间：2024.10.03
    环境：CartPole 
    作者：不去幼儿园
"""
import gym
import numpy as np
import random

# 定义超参数
GAMMA = 0.99
LEARNING_RATE = 0.1
EPSILON_DECAY = 0.995
MIN_EPSILON = 0.1
NUM_EPISODES = 500

# 定义 MAXQ 子任务节点
class MAXQNode:
    def __init__(self, num_actions, is_primitive=False):
        self.num_actions = num_actions
        self.is_primitive = is_primitive  # 是否为原子任务
        self.q_values = {}  # Q 值字典，(state, action) -> Q 值

    def get_q(self, state, action):
        state = tuple(state)  # 将 state 转换为元组
        if (state, action) not in self.q_values:
            self.q_values[(state, action)] = 0.0  # 初始化 Q 值
        return self.q_values[(state, action)]

    def set_q(self, state, action, value):
        state = tuple(state)  # 将 state 转换为元组
        self.q_values[(state, action)] = value

# 创建 MAXQ 分解的顶层任务
class MAXQTask:
    def __init__(self, num_actions):
        self.num_actions = num_actions
        self.subtasks = []
        self.root = MAXQNode(num_actions)  # 顶层任务节点

    def add_subtask(self, subtask):
        self.subtasks.append(subtask)

# 定义智能体
class MAXQAgent:
    def __init__(self, env):
        self.env = env
        self.epsilon = 1.0
        self.maxq_root = MAXQTask(env.action_space.n)  # CartPole 的动作数为 2

        # 添加子任务
        self.pickup_subtask = MAXQNode(env.action_space.n, is_primitive=True)  # 装载任务
        self.dropoff_subtask = MAXQNode(env.action_space.n, is_primitive=True)  # 卸载任务
        self.navigate_subtask = MAXQNode(env.action_space.n)  # 导航任务

        # 将子任务添加到根节点
        self.maxq_root.add_subtask(self.pickup_subtask)
        self.maxq_root.add_subtask(self.dropoff_subtask)
        self.maxq_root.add_subtask(self.navigate_subtask)

    def select_action(self, state, subtask, epsilon):
        # ε-贪婪策略选择动作
        if random.random() < epsilon:
            return self.env.action_space.sample()  # 探索
        else:
            q_values = [subtask.get_q(state, action) for action in range(subtask.num_actions)]
            return np.argmax(q_values)  # 利用当前策略

    def update_q(self, state, subtask, action, reward, next_state):
        # 确保传入的是 MAXQNode 而不是 MAXQTask
        if not isinstance(subtask, MAXQNode):
            subtask = subtask.root  # 传入的是 MAXQTask，则使用其根节点
        next_q_values = [subtask.get_q(next_state, a) for a in range(subtask.num_actions)]
        max_next_q = max(next_q_values)

        # Q 值更新
        current_q = subtask.get_q(state, action)
        new_q = current_q + LEARNING_RATE * (reward + GAMMA * max_next_q - current_q)
        subtask.set_q(state, action, new_q)

    def train(self, num_episodes):
        for episode in range(num_episodes):
            state,_ = self.env.reset()
            done = False
            total_reward = 0

            while not done:
                # 确保传递的是 MAXQNode
                action = self.select_action(state, self.maxq_root.root, self.epsilon)
                next_state, reward, done, _, __ = self.env.step(action)

                # 这里传递的是 MAXQNode，而不是 MAXQTask
                self.update_q(state, self.maxq_root.root, action, reward, next_state)
                state = next_state
                total_reward += reward

            # 逐渐减少 epsilon 以减少探索
            self.epsilon = max(MIN_EPSILON, self.epsilon * EPSILON_DECAY)
            print(f"Episode {episode + 1}: Total Reward: {total_reward}")

# 创建 CartPole 环境并训练智能体
env = gym.make('CartPole-v1')
agent = MAXQAgent(env)
agent.train(NUM_EPISODES)
env.close()

测试代码实现：

import gym
import torch

# 测试 MAXQ 智能体并显示动画
def test_maxq_agent(agent, env, num_episodes=5):
    for episode in range(num_episodes):
        state, _ = env.reset()
        done = False
        total_reward = 0
        env.render()  # 初始化渲染

        while not done:
            env.render()  # 显示动画
            action = agent.select_action(state, agent.maxq_root.root, epsilon=0.0)  # 使用已学策略选择动作
            next_state, reward, done, _, __ = env.step(action)
            state = next_state
            total_reward += reward

        print(f"测试 Episode {episode + 1}: Total Reward: {total_reward}")
    env.close()

# 创建环境并调用测试函数
env = gym.make('CartPole-v1', render_mode='human')
test_maxq_agent(agent, env)