本文目录导读：
我来详细讲解数据分层强化学习的Python实现拆解方法。
基础框架搭建
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque
import random
class HRLFramework:
"""分层强化学习基础框架"""
def __init__(self, state_dim, action_dim, high_level_dim=10, low_level_dim=5):
self.high_level_controller = HighLevelPolicy(state_dim, high_level_dim)
self.low_level_controller = LowLevelPolicy(state_dim, action_dim, high_level_dim)
self.meta_controller = MetaController(state_dim, high_level_dim)
def get_action(self, state, goal=None):
if goal is None:
goal = self.meta_controller.select_goal(state)
action = self.low_level_controller.get_action(state, goal)
return action, goal
拆解步骤详解
1 目标生成层 (High-Level)
class HighLevelPolicy(nn.Module):
"""高层策略: 生成子目标和抽象行为"""
def __init__(self, state_dim, goal_dim):
super().__init__()
self.network = nn.Sequential(
nn.Linear(state_dim, 128),
nn.ReLU(),
nn.Linear(128, 64),
nn.ReLU(),
nn.Linear(64, goal_dim),
nn.Tanh() # 输出归一化到[-1, 1]
)
def forward(self, state):
return self.network(state)
def select_goal(self, state, n_steps=10):
"""生成n步内要达成的子目标"""
goal = self.forward(state)
return goal
2 底层执行层 (Low-Level)
class LowLevelPolicy(nn.Module):
"""底层策略: 根据子目标执行具体动作"""
def __init__(self, state_dim, action_dim, goal_dim):
super().__init__()
self.network = nn.Sequential(
nn.Linear(state_dim + goal_dim, 256),
nn.ReLU(),
nn.Linear(256, 128),
nn.ReLU(),
nn.Linear(128, action_dim),
nn.Tanh()
)
def forward(self, state, goal):
# 将状态和目标拼接
state_goal = torch.cat([state, goal], dim=-1)
return self.network(state_goal)
def get_action(self, state, goal, epsilon=0.1):
if random.random() < epsilon:
return torch.randn(state.shape[0], self.action_dim) # 探索
else:
return self.forward(state, goal) # 利用
3 元控制器 (Meta-Controller)
class MetaController:
"""元控制器: 管理高层和低层的交互"""
def __init__(self, state_dim, goal_dim, max_substeps=50):
self.goal_dim = goal_dim
self.max_substeps = max_substeps
self.goal_history = deque(maxlen=10)
def select_goal(self, state, high_level_policy):
"""根据当前状态选择新目标"""
# 检查是否需要更新目标
if self._should_update_goal(state):
new_goal = high_level_policy(state)
self.goal_history.append(new_goal)
return new_goal
return self.goal_history[-1] if self.goal_history else None
def _should_update_goal(self, state):
"""判断是否需要更新目标"""
# 可以根据时间步、状态变化等条件判断
return len(self.goal_history) == 0 or random.random() < 0.1
def check_goal_achieved(self, state, goal, threshold=0.1):
"""检查目标是否达成"""
distance = torch.norm(state - goal, dim=-1)
return distance < threshold
4 分层经验回放
class HierarchicalReplayBuffer:
"""分层经验回放池"""
def __init__(self, capacity=100000):
self.capacity = capacity
self.high_level_buffer = deque(maxlen=capacity)
self.low_level_buffer = deque(maxlen=capacity)
def add_high_level_experience(self, state, goal, reward, next_state, done):
"""添加高层经验"""
self.high_level_buffer.append({
'state': state,
'goal': goal,
'reward': reward,
'next_state': next_state,
'done': done
})
def add_low_level_experience(self, state, goal, action, reward, next_state, done):
"""添加低层经验"""
self.low_level_buffer.append({
'state': state,
'goal': goal,
'action': action,
'reward': reward,
'next_state': next_state,
'done': done
})
def sample(self, batch_size, level='low'):
"""采样经验"""
buffer = self.low_level_buffer if level == 'low' else self.high_level_buffer
batch = random.sample(buffer, min(len(buffer), batch_size))
return batch
完整训练流程
class HierarchicalTrainer:
"""分层强化学习训练器"""
def __init__(self, env, high_lr=3e-4, low_lr=1e-3, gamma=0.99):
self.env = env
self.gamma = gamma
# 初始化网络
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.shape[0]
goal_dim = 10
self.high_policy = HighLevelPolicy(state_dim, goal_dim)
self.low_policy = LowLevelPolicy(state_dim, action_dim, goal_dim)
self.meta_controller = MetaController(state_dim, goal_dim)
self.buffer = HierarchicalReplayBuffer()
# 优化器
self.high_optimizer = optim.Adam(self.high_policy.parameters(), lr=high_lr)
self.low_optimizer = optim.Adam(self.low_policy.parameters(), lr=low_lr)
def train_episode(self):
"""训练一个episode"""
state = self.env.reset()
episode_reward = 0
done = False
step = 0
# 获取初始目标
goal = self.meta_controller.select_goal(
torch.FloatTensor(state), self.high_policy
)
while not done:
# 低层执行
action = self.low_policy.get_action(
torch.FloatTensor(state), torch.FloatTensor(goal)
)
next_state, reward, done, _ = self.env.step(action.detach().numpy())
# 检查目标是否达成
goal_achieved = self.meta_controller.check_goal_achieved(
torch.FloatTensor(next_state),
torch.FloatTensor(goal)
)
# 计算内部奖励（目标导向）
intrinsic_reward = self._compute_intrinsic_reward(state, goal, next_state)
# 存储低层经验
self.buffer.add_low_level_experience(
state, goal, action, intrinsic_reward, next_state, done
)
# 更新目标
if goal_achieved or step % 10 == 0:
# 存储高层经验
high_reward = reward if goal_achieved else -1
self.buffer.add_high_level_experience(
state, goal, high_reward, next_state, done
)
# 选择新目标
goal = self.meta_controller.select_goal(
torch.FloatTensor(next_state), self.high_policy
)
state = next_state
episode_reward += reward
step += 1
# 更新网络
if len(self.buffer.low_level_buffer) > 100:
self._update_networks()
return episode_reward
def _compute_intrinsic_reward(self, state, goal, next_state):
"""计算内部奖励（基于目标达成程度）"""
current_dist = torch.norm(torch.FloatTensor(state) - torch.FloatTensor(goal))
next_dist = torch.norm(torch.FloatTensor(next_state) - torch.FloatTensor(goal))
return current_dist - next_dist # 距离减少为正奖励
def _update_networks(self, batch_size=64):
"""更新网络参数"""
# 更新低层
low_batch = self.buffer.sample(batch_size, level='low')
self._update_low_level(low_batch)
# 更新高层
high_batch = self.buffer.sample(batch_size, level='high')
self._update_high_level(high_batch)
def _update_low_level(self, batch):
"""更新低层策略"""
states = torch.FloatTensor([e['state'] for e in batch])
goals = torch.FloatTensor([e['goal'] for e in batch])
actions = torch.FloatTensor([e['action'] for e in batch])
rewards = torch.FloatTensor([e['reward'] for e in batch])
next_states = torch.FloatTensor([e['next_state'] for e in batch])
dones = torch.FloatTensor([e['done'] for e in batch])
# 计算TD误差
with torch.no_grad():
next_actions = self.low_policy(next_states, goals)
# 这里可以添加Q网络计算目标值
# 更新策略（简化版本）
self.low_optimizer.zero_grad()
predicted_actions = self.low_policy(states, goals)
loss = nn.MSELoss()(predicted_actions, actions)
loss.backward()
self.low_optimizer.step()
def _update_high_level(self, batch):
"""更新高层策略"""
states = torch.FloatTensor([e['state'] for e in batch])
goals = torch.FloatTensor([e['goal'] for e in batch])
rewards = torch.FloatTensor([e['reward'] for e in batch])
# 更新高层策略（简化版本）
self.high_optimizer.zero_grad()
predicted_goals = self.high_policy(states)
loss = nn.MSELoss()(predicted_goals, goals)
loss.backward()
self.high_optimizer.step()
关键拆解技巧
1 时间尺度解耦
class TimeScaleDecoupling:
"""时间尺度解耦"""
def __init__(self, high_freq=10, low_freq=1):
self.high_step = 0
self.high_freq = high_freq
self.low_freq = low_freq
def should_update_high(self, step):
"""判断是否更新高层"""
return step % self.high_freq == 0
def should_update_low(self, step):
"""判断是否更新低层"""
return step % self.low_freq == 0
2 奖励分解
class RewardDecomposition:
"""奖励分解"""
def __init__(self, extrinsic_weight=1.0, intrinsic_weight=0.1):
self.extrinsic_weight = extrinsic_weight
self.intrinsic_weight = intrinsic_weight
def decompose_reward(self, state, action, next_state, goal):
"""分解奖励为外部和内部"""
extrinsic_reward = self._get_extrinsic_reward(state, action, next_state)
intrinsic_reward = self._get_intrinsic_reward(state, goal, next_state)
return {
'extrinsic': extrinsic_reward * self.extrinsic_weight,
'intrinsic': intrinsic_reward * self.intrinsic_weight,
'total': extrinsic_reward * self.extrinsic_weight + \
intrinsic_reward * self.intrinsic_weight
}
def _get_extrinsic_reward(self, state, action, next_state):
"""外部环境奖励"""
# 根据具体环境实现
pass
def _get_intrinsic_reward(self, state, goal, next_state):
"""内部探索奖励"""
goal_progress = np.linalg.norm(state - goal) - np.linalg.norm(next_state - goal)
return max(0, goal_progress)
3 状态抽象
class StateAbstraction:
"""状态抽象层"""
def __init__(self, input_dim, abstract_dim=32):
self.encoder = nn.Sequential(
nn.Linear(input_dim, 64),
nn.ReLU(),
nn.Linear(64, abstract_dim)
)
self.decoder = nn.Sequential(
nn.Linear(abstract_dim, 64),
nn.ReLU(),
nn.Linear(64, input_dim)
)
def abstract_state(self, state):
"""抽象状态表示"""
with torch.no_grad():
return self.encoder(state)
def reconstruct_state(self, abstract_state):
"""从抽象状态重建原始状态"""
return self.decoder(abstract_state)
实战示例：导航任务
class NavigationHRL:
"""导航任务的分层强化学习"""
def __init__(self, env):
self.env = env
self.setup_hierarchical_policies()
def setup_hierarchical_policies(self):
# 高层：选择目标位置
self.high_level = {
'policy': self._select_waypoint,
'reward': self._compute_waypoint_reward,
'horizon': 20 # 每20步更新一次目标
}
# 低层：执行运动控制
self.low_level = {
'policy': self._move_to_waypoint,
'reward': self._compute_movement_reward,
'horizon': 1 # 每一步都执行
}
def _select_waypoint(self, state):
"""选择下一个路径点"""
current_pos = state[:2]
target_pos = state[2:4]
# 计算中间路径点
waypoint = current_pos + (target_pos - current_pos) * 0.3
return waypoint
def _move_to_waypoint(self, state, waypoint):
"""移动到路径点"""
current_pos = state[:2]
direction = waypoint - current_pos
# 归一化并输出动作
action = direction / (np.linalg.norm(direction) + 1e-8)
return action
def train(self, episodes=1000):
"""训练导航任务"""
for episode in range(episodes):
state = self.env.reset()
episode_reward = 0
done = False
waypoint_update_counter = 0
# 初始化路径点
current_waypoint = self._select_waypoint(state)
while not done:
# 检查是否需要更新路径点
if waypoint_update_counter % self.high_level['horizon'] == 0:
current_waypoint = self._select_waypoint(state)
# 执行低层动作
action = self._move_to_waypoint(state, current_waypoint)
next_state, reward, done, _ = self.env.step(action)
# 计算分层奖励
high_reward = self.high_level['reward'](state, current_waypoint, next_state)
low_reward = self.low_level['reward'](state, action, next_state, current_waypoint)
episode_reward += reward
state = next_state
waypoint_update_counter += 1
print(f"Episode {episode}: Reward = {episode_reward}")
这个拆解方法的核心思想是将复杂任务分解为多个抽象层次,每层负责不同粒度的决策，通过层间交互实现端到端的学习。
