【RL Latest Tech】分层强化学习：FeUdal Networks算法

FeUdal Networks（FuN）是一种分层强化学习（Hierarchical Reinforcement Learning, HRL）算法，由Google DeepMind团队提出。该算法的灵感来源于层级控制结构，将任务分解为高层目标和低层执行细节，从而提高强化学习在复杂环境中的效率。与传统的强化学习算法不同，FeUdal Networks将学习过程分为不同的层次，每个层次的角色不同，但都

不去幼儿园

1586人浏览 · 2024-10-05 11:30:00

不去幼儿园 · 2024-10-05 11:30:00 发布

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

【强化学习】（24）---《分层强化学习：FeUdal Networks算法》

分层强化学习：FeUdal Networks算法

1. FeUdal Networks的核心思想

2. 结构与机制

(1) 管理者网络（Manager Network）

(2) 工人网络（Worker Network）

(3) 内在奖励信号

(4) 全局奖励更新

3. FeUdal Networks的工作流程

[Python] Feudal Netwrks算法实现

5. FeUdal Networks的应用场景

6. 相关工作

FeUdal Networks（FuN）是一种分层强化学习（Hierarchical Reinforcement Learning, HRL）算法，由Google DeepMind团队提出。该算法的灵感来源于层级控制结构，将任务分解为高层目标和低层执行细节，从而提高强化学习在复杂环境中的效率。与传统的强化学习算法不同，FeUdal Networks将学习过程分为不同的层次，每个层次的角色不同，但都为共同完成任务服务。

以下是FeUdal Networks算法的详细介绍：

1. FeUdal Networks的核心思想

FeUdal Networks的核心思想是将强化学习任务划分为高层策略（manager）和低层策略（worker）：

Manager（管理者）：高层策略负责决定任务的总体目标，通常以更高的抽象层次进行决策。它在高维的状态空间中学习如何为低层策略设定子目标（sub-goals）。
Worker（工人）：低层策略负责在具体的动作空间中实现高层设定的子目标，即执行实际的动作序列来完成高层指定的目标。

通过这种分层结构，FeUdal Networks可以将复杂任务逐步简化，提升学习效率。

2. 结构与机制

FeUdal Networks的网络结构分为两个部分：管理者网络（Manager Network）和工人网络（Worker Network）。它们各自负责不同的功能，协同完成整个任务。

(1) 管理者网络（Manager Network）

管理者网络负责为工人网络设定高层次的目标。具体来说，它会在较长时间尺度上学习如何选择目标（goals），这些目标通常是从状态空间中的某个子空间中抽取出来的。管理者通过学习如何在这些高维空间中生成有意义的目标，来指导工人如何执行。

管理者每隔固定的时间步长（称为时间地平线，如10步或20步）生成一个新的子目标。这个子目标表示工人接下来的一段时间内需要达到的状态或方向。子目标通常是一个在状态空间中的矢量。

管理者目标生成

设定管理者生成的目标为一个向量，表示为：

$[ g_t = f_\theta(s_t) ]$

其中：

$(g_t)$ 是在时间步 $(t)$ 时生成的目标向量。
$(f_\theta(s_t))$ 是管理者网络通过状态(s_t)生成目标的函数。

管理者的目标是引导工人在时间地平线内（例如10步内）完成该目标。

(2) 工人网络（Worker Network）

工人网络在时间地平线内负责根据管理者设定的目标采取实际的行动。工人从管理者网络处接收到一个目标矢量，并将其用作输入来确定当前需要采取的具体动作。

工人网络直接与环境交互，通过动作空间中的操作来逐步朝着管理者设定的目标前进。工人网络学习的奖励来自环境的即时反馈，同时工人要学会如何在管理者设定的目标指导下实现最优的动作序列。

工人网络的行为策略依赖于管理者提供的目标向量和当前状态的组合，工人选择一个动作 $(a_t)$ ，具体可以表示为：

$[ a_t \sim \pi(a_t | s_t, g_{t-1}) ]$

其中：

$(a_t)$ 是在时间步 $(t)$ 时工人选择的动作。
$(\pi(a_t | s_t, g_{t-1}))$ 是基于当前状态(s_t)和管理者在之前时间步生成的目标向量 $(g_{t-1})$ 的策略。

(3) 内在奖励信号

为了增强学习的效果，FeUdal Networks引入了内在奖励信号。与标准强化学习不同的是，工人网络并不直接从环境的全局奖励中学习，而是从管理者提供的内在奖励中学习。内在奖励由工人是否实现了管理者指定的目标决定。如果工人成功靠近管理者指定的目标位置，工人将获得高的内在奖励，从而优化其行为。

管理者网络则从全局环境的反馈（全局奖励）中学习如何选择有效的子目标，以使工人能够更好地完成任务。

工人的奖励并不直接来自环境，而是通过内在奖励信号计算得出。内在奖励信号是工人实现管理者目标的程度，具体计算方式为：

$[ r_t^{\text{intrinsic}} = g_{t-1} \cdot (s_t - s_{t-1}) ]$

其中：

$(r_t^{\text{intrinsic}})$ 是工人在时间步 $(t)$ 获得的内在奖励。
$(g_{t-1})$ 是管理者在时间步 $(t-1)$ 设定的目标向量。
$(s_t - s_{t-1})$ 是工人在时间步 $(t)$ 时状态的变化量。

该内在奖励通过计算目标向量与状态变化量的内积来判断工人是否朝着管理者设定的方向前进。

(4) 全局奖励更新

尽管工人网络通过内在奖励进行学习，管理者网络仍然通过全局环境的奖励信号进行更新，优化目标生成策略。管理者的学习目标是最大化未来期望奖励：

$[ R = \sum_{t=0}^{T} \gamma^t r_t ]$

其中：

$(R)$ 是未来期望奖励的总和。
$(r_t)$ 是时间步 $(t)$ 获得的全局奖励信号。
$(\gamma)$ 是折扣因子，用于控制短期和长期奖励的权衡。

3. FeUdal Networks的工作流程

管理者生成目标：管理者网络在时间地平线到达时生成一个子目标，目标通常是状态空间的一个方向或位置矢量。
工人执行动作：工人网络根据当前的环境状态和管理者的目标，选择具体的动作序列，逐步接近管理者设定的子目标。
工人获得内在奖励：工人网络通过内在奖励函数来评估自己是否成功完成了子目标，并根据这一奖励调整策略。
管理者更新目标：在每一个时间地平线结束时，管理者网络会更新其策略，以便在后续的时间步长中生成更有效的子目标。
全局反馈：管理者根据全局环境的奖励反馈，学习如何生成更合理的子目标。

[Python] Feudal Netwrks算法实现

Feudal Networks (FeUdal Networks) 是一种分层强化学习方法，它通过高层（manager）和低层（worker）的相互作用来学习复杂任务。高层决定子目标，低层根据子目标学习策略。下面是 FeUdal Networks 算法在 CartPole 环境中的 Python 实现，同时包含测试代码和动画显示。

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

主要步骤：

高层（manager）：设定子目标方向，通常是一些隐空间特征。
低层（worker）：根据高层给定的子目标来执行动作，并学习策略。
学习和更新：高层和低层都有各自的策略和目标，它们通过相互作用来优化整体策略。

环境要求：

安装 gym：pip install gym
安装 torch：pip install torch
安装 pygame（用于动画显示）：pip install pygame

`算法训练代码:`

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

# 超参数
GAMMA = 0.99
LEARNING_RATE = 0.001
MANAGER_UPDATE_FREQUENCY = 10  # 高层更新频率
WORKER_UPDATE_FREQUENCY = 1  # 低层更新频率
NUM_EPISODES = 500
MIN_EPSILON = 0.1
EPSILON_DECAY = 0.995


# 高层（Manager）网络
class ManagerNetwork(nn.Module):
	def __init__(self, state_dim, goal_dim):
		super(ManagerNetwork, self).__init__()
		self.fc1 = nn.Linear(state_dim, 128)
		self.fc2 = nn.Linear(128, goal_dim)

	def forward(self, state):
		x = torch.relu(self.fc1(state))
		goal = self.fc2(x)
		return goal


# 低层（Worker）网络
class WorkerNetwork(nn.Module):
	def __init__(self, state_dim, goal_dim, action_dim):
		super(WorkerNetwork, self).__init__()
		self.fc1 = nn.Linear(state_dim + goal_dim, 128)
		self.fc2 = nn.Linear(128, action_dim)

	def forward(self, state, goal):
		x = torch.cat((state, goal), dim=-1)
		x = torch.relu(self.fc1(x))
		action_logits = self.fc2(x)
		return action_logits


# FeUdal Networks 智能体
class FeudalAgent:
	def __init__(self, state_dim, action_dim, goal_dim):
		self.manager_net = ManagerNetwork(state_dim, goal_dim)
		self.worker_net = WorkerNetwork(state_dim, goal_dim, action_dim)
		self.manager_optimizer = optim.Adam(self.manager_net.parameters(), lr=LEARNING_RATE)
		self.worker_optimizer = optim.Adam(self.worker_net.parameters(), lr=LEARNING_RATE)
		self.epsilon = 1.0

	def select_worker_action(self, state, goal, epsilon):
		if random.random() < epsilon:
			return random.choice([0, 1])  # CartPole 动作空间为 2（0 或 1）
		else:
			state = torch.FloatTensor(state).unsqueeze(0)
			goal = torch.FloatTensor(goal).unsqueeze(0)
			action_logits = self.worker_net(state, goal)
			action_probs = torch.softmax(action_logits, dim=-1)
			return torch.argmax(action_probs).item()

	# 修改后的 update_manager 方法
	def update_manager(self, state, next_state, goal, reward):
		state = torch.FloatTensor(state).unsqueeze(0).requires_grad_(True)  # 启用 requires_grad
		next_state = torch.FloatTensor(next_state).unsqueeze(0).requires_grad_(True)  # 启用 requires_grad
		goal = torch.FloatTensor(goal).unsqueeze(0).requires_grad_(True)  # 启用 requires_grad

		# 计算内在奖励
		intrinsic_reward = torch.dot(goal.squeeze(), next_state.squeeze() - state.squeeze())

		# 反向传播损失
		loss = -intrinsic_reward * reward
		self.manager_optimizer.zero_grad()
		loss.backward()
		self.manager_optimizer.step()

	def update_worker(self, state, goal, action, reward):
		state = torch.FloatTensor(state).unsqueeze(0)
		goal = torch.FloatTensor(goal).unsqueeze(0)
		action_logits = self.worker_net(state, goal)
		action_probs = torch.softmax(action_logits, dim=-1)
		log_action_probs = torch.log(action_probs)
		policy_loss = -log_action_probs[0, action] * reward
		self.worker_optimizer.zero_grad()
		policy_loss.backward()
		self.worker_optimizer.step()

	def train(self, env, num_episodes):
		goal_dim = self.manager_net.fc2.out_features

		for episode in range(num_episodes):
			state, _ = env.reset()
			goal = self.manager_net(torch.FloatTensor(state).unsqueeze(0)).detach().numpy().squeeze()
			done = False
			episode_reward = 0
			steps = 0

			while not done:
				steps += 1
				action = self.select_worker_action(state, goal, self.epsilon)
				next_state, reward, done, _, __ = env.step(action)

				# 更新低层（Worker）
				self.update_worker(state, goal, action, reward)

				# 每隔 MANAGER_UPDATE_FREQUENCY 更新一次高层（Manager）
				if steps % MANAGER_UPDATE_FREQUENCY == 0:
					new_goal = self.manager_net(torch.FloatTensor(next_state).unsqueeze(0)).detach().numpy().squeeze()
					self.update_manager(state, next_state, goal, reward)
					goal = new_goal

				state = next_state
				episode_reward += reward

			self.epsilon = max(MIN_EPSILON, self.epsilon * EPSILON_DECAY)
			print(f"Episode {episode + 1}: Total Reward: {episode_reward}")


# 创建 CartPole 环境并训练智能体
env = gym.make('CartPole-v1')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
goal_dim = 4  # 设置一个目标维度
agent = FeudalAgent(state_dim, action_dim, goal_dim)
agent.train(env, NUM_EPISODES)

`算法测试代码:`

# 测试 FeUdal Networks 智能体
def test_feudal_agent(agent, env, num_episodes=5):
	for episode in range(num_episodes):
		state, _ = env.reset()
		goal = agent.manager_net(torch.FloatTensor(state).unsqueeze(0)).detach().numpy().squeeze()
		done = False
		total_reward = 0
		env.render()

		while not done:
			env.render()
			action = agent.select_worker_action(state, goal, epsilon=0.0)
			next_state, reward, done, _, __ = env.step(action)
			state = next_state
			total_reward += reward

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


# 测试智能体并显示动画
env = gym.make('CartPole-v1', render_mode='human')
test_feudal_agent(agent, env)

[Notice] 注意事项

渲染模式：使用 gym.make('CartPole-v1', render_mode='human') 来显示动画。
环境：确保在本地 Python 环境中运行，并安装了所有依赖库。
内在奖励：计算内在奖励时，使用高层设定的目标来激励低层智能体。

说明：

高层网络（ManagerNetwork）：根据当前状态设定一个目标（子目标），通常为一个隐空间中的向量。
低层网络（WorkerNetwork）：根据状态和高层设定的目标来选择动作。
高层更新：每隔一段时间更新高层的目标，并计算内在奖励（intrinsic reward）。
低层更新：低层根据高层设定的目标执行动作，并利用策略梯度方法更新策略。
测试代码：在 test_feudal_agent 函数中，使用训练后的策略来执行 CartPole 环境中的测试，并显示动画。

env = gym.make("Taxi-v3", render_mode="human")  # 使用人类可视化的方式

调整参数：

可以调整 goal_dim 来改变高层目标的复杂性。
调整 MANAGER_UPDATE_FREQUENCY 和 WORKER_UPDATE_FREQUENCY 以改变高层和低层之间的更新频率。

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

4. 优点与挑战

优点：

层次结构提升学习效率：通过将复杂任务分解为目标设定和动作执行，FeUdal Networks可以更高效地学习解决复杂的任务，特别是在具有长期依赖性的任务中表现优异。
内在奖励机制：工人网络通过内在奖励进行优化，减少了全局奖励稀疏的挑战，提高了学习的稳定性。
时间尺度分离：管理者和工人在不同的时间尺度上进行学习和优化，使得算法能够更好地应对长期决策问题。

挑战：

目标空间设计：如何设计合适的子目标空间，以及确保目标能够被有效达成，是算法实现中的关键挑战。
任务分解难度：虽然算法可以分解任务，但对于不同类型的任务，任务分解的有效性可能有所不同。
时间地平线选择：不同的时间地平线对学习效果的影响较大，选择合适的时间地平线需要一定的经验和调试。

5. FeUdal Networks的应用场景

FeUdal Networks适用于那些任务具有层次结构的场景，尤其是在以下情况下表现优异：

长时间跨度的任务：例如，在策略游戏、机器人导航等任务中，任务往往需要在多个时间步长上进行连续的决策。
稀疏奖励问题：对于环境中奖励信号稀疏的任务，FeUdal Networks通过内在奖励机制，可以有效缓解学习困难。
多级目标决策：需要多级目标设定和执行的场景，如复杂的机器人控制或多步骤规划任务。

6. 相关工作

FeUdal Networks是分层强化学习领域的一项重要进展，它与其他分层学习方法（如选项框架（Options Framework）和MAXQ分解算法）具有相似之处，但更加关注通过管理者和工人的分工合作来处理不同的时间尺度。FeUdal Networks在大型复杂任务中的表现使其成为近年来强化学习领域的重要研究方向之一。