Python脚本数据分层强化学习如何拆解

wen 实用脚本 2026-07-18 170

本文目录导读：

基础框架搭建
拆解步骤详解
完整训练流程
关键拆解技巧
实战示例：导航任务

我来详细讲解数据分层强化学习的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}")

这个拆解方法的核心思想是将复杂任务分解为多个抽象层次,每层负责不同粒度的决策，通过层间交互实现端到端的学习。

本文地址： https://www.bmjup.com/post/24142.html