百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术分类 > 正文

PyTorch 深度学习实战(26):多目标强化学习Multi-Objective RL

ztj100 2025-04-26 22:45 168 浏览 0 评论

一、多目标强化学习原理

1. 多目标学习核心思想

多目标强化学习(Multi-Objective RL)旨在让智能体同时优化多个冲突目标,通过平衡目标间的权衡关系找到帕累托最优解集。与传统强化学习的区别在于:

对比维度

传统强化学习

多目标强化学习

目标数量

单一奖励函数

多个奖励函数(可能相互冲突)

优化目标

最大化单一累计奖励

找到帕累托最优策略集合

解的唯一性

唯一最优解

多个非支配解(Pareto Front)

应用场景

目标明确且无冲突的任务

自动驾驶(安全 vs 效率)、资源分配

2. 多目标问题建模


二、标量化方法(Scalarization)


三、多目标 PPO 算法实现(基于 Gymnasium)

自定义机器人控制环境 为例,实现基于权重调整的多目标 PPO 算法:

  1. 定义多目标环境:机器人需同时最大化前进速度和最小化能耗
  2. 构建策略网络:共享特征提取层 + 多目标价值头
  3. 动态权重调整:根据训练阶段调整目标权重
  4. 帕累托前沿分析:评估不同权重下的策略性能

四、代码实现

Bash
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from torch.distributions import Normal
import gymnasium as gym
import time

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# ================== 自定义多目标环境 ==================
class MultiObjectiveRobotEnv(gym.Env):
    def __init__(self):
        self.observation_space = gym.spaces.Box(low=-1, high=1, shape=(4,))
        self.action_space = gym.spaces.Box(low=-1, high=1, shape=(2,))
        self.state = None
        self.steps = 0
        self.max_steps = 200  # 添加最大步数限制
    
    def reset(self):
        self.state = np.random.uniform(-1, 1, size=(4,))
        self.steps = 0
        return self.state.copy()
    
    def step(self, action):
        # Clip actions to valid range
        action = np.clip(action, -1, 1)
        
        # 定义两个冲突目标:速度(正向奖励)和能耗(负向奖励)
        velocity_reward = np.abs(action[0])  # 速度与动作绝对值正相关
        energy_cost = 0.1 * np.sum(np.square(action))  # 能耗与动作平方正相关
        
        self.state += 0.1 * np.array([action[0], action[1], 0, 0])  # 简化动力学
        self.state = np.clip(self.state, -1, 1)  # 保持状态在合理范围内
        
        self.steps += 1
        done = self.steps >= self.max_steps  # 超过最大步数时终止
        
        return self.state.copy(), [velocity_reward, -energy_cost], done, {}

# ================== 多目标 PPO 策略网络 ==================
class MultiObjectivePPO(nn.Module):
    def __init__(self, state_dim, action_dim, num_objectives=2):
        super().__init__()
        # 共享特征提取层
        self.shared_net = nn.Sequential(
            nn.Linear(state_dim, 256),
            nn.Tanh(),  # Using Tanh for more stable gradients
            nn.Linear(256, 256),
            nn.Tanh()
        )
        # 多目标价值头
        self.value_heads = nn.ModuleList([
            nn.Sequential(nn.Linear(256, 1), nn.Tanh()) for _ in range(num_objectives)
        ])
        # 策略头
        self.actor_mean = nn.Sequential(
            nn.Linear(256, action_dim),
            nn.Tanh()  # Output between -1 and 1
        )
        self.actor_log_std = nn.Parameter(torch.zeros(action_dim) - 1.0)  # Initialize to smaller std
    
    def forward(self, state):
        features = self.shared_net(state)
        values = [head(features) for head in self.value_heads]
        action_mean = self.actor_mean(features)
        action_std = torch.exp(self.actor_log_std).clamp(1e-4, 1.0)  # Clamp std to avoid NaN
        return action_mean, action_std, values

# ================== 训练系统 ==================
class MultiObjectivePPOTrainer:
    def __init__(self):
        self.env = MultiObjectiveRobotEnv()
        self.state_dim = self.env.observation_space.shape[0]
        self.action_dim = self.env.action_space.shape[0]
        self.num_objectives = 2
        self.policy = MultiObjectivePPO(self.state_dim, self.action_dim).to(device)
        self.optimizer = optim.Adam(self.policy.parameters(), lr=1e-4)  # Reduced learning rate
        self.weights = np.array([0.5, 0.5])  # 初始权重
        self.gamma = 0.99  # Discount factor
        self.clip_epsilon = 0.2  # PPO clip parameter
        self.max_steps_per_episode = 200  # 每episode最大步数
    
    def update_weights(self, episode):
        # 动态调整权重(示例:周期变化)
        self.weights = np.array([np.sin(episode * 0.01) * 0.5 + 0.5, 
                                1 - (np.sin(episode * 0.01) * 0.5 + 0.5)])
        self.weights /= np.sum(self.weights)
    
    def train(self, max_episodes=1000):
        for episode in range(max_episodes):
            self.update_weights(episode)
            state = self.env.reset()
            episode_rewards = np.zeros(self.num_objectives)
            episode_steps = 0
            
            # 采集轨迹数据
            states, actions, log_probs, objectives = [], [], [], []
            
            for _ in range(self.max_steps_per_episode):
                state_tensor = torch.FloatTensor(state).unsqueeze(0).to(device)
                with torch.no_grad():
                    action_mean, action_std, values = self.policy(state_tensor)
                    dist = Normal(action_mean, action_std)
                    action = dist.sample()
                    log_prob = dist.log_prob(action).sum(dim=-1)
                
                next_state, obj_rewards, done, _ = self.env.step(action.squeeze(0).cpu().numpy())
                
                states.append(state)
                actions.append(action.squeeze(0))
                log_probs.append(log_prob)
                objectives.append(obj_rewards)
                episode_rewards += np.array(obj_rewards)
                episode_steps += 1
                state = next_state
                
                if done:
                    break
            
            # 计算加权累计奖励
            weighted_reward = np.dot(episode_rewards, self.weights)
            if (episode + 1) % 100 == 0:
                print(f"Episode {episode+1}/{max_episodes} | Steps: {episode_steps} | "
                    f"Weight: {self.weights} | Reward: {weighted_reward:.1f} | "
                    f"Obj1: {episode_rewards[0]:.1f} | Obj2: {episode_rewards[1]:.1f}")

if __name__ == "__main__":
    start = time.time()
    start_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(start))
    print(f"开始时间: {start_str}")
    print("初始化环境...")
    trainer = MultiObjectivePPOTrainer()
    trainer.train(max_episodes=500)  # 先测试少量episode
    end = time.time()
    end_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(end))
    print(f"训练完成时间: {end_str}")
    print(f"训练完成,耗时: {end - start:.2f}秒")

五、关键代码解析

1.多目标环境设计

  • step() 返回两个奖励:速度奖励(velocity_reward)和能耗惩罚(-energy_cost)。
  • 通过调整动作的绝对值(速度)和平方值(能耗)实现目标冲突。

2.动态权重调整

  • update_weights() 周期性调整目标权重(示例中使用正弦函数)。
  • 实际应用中可根据需求设计自适应权重策略。

3.策略网络结构

  • 共享特征提取层(shared_net)学习状态共性表示。
  • 独立价值头(value_heads)分别预测各目标的价值。

六、训练输出示例

Bash
开始时间: 2025-03-26 03:37:15
初始化环境...
Episode 100/500 | Steps: 200 | Weight: [0.91801299 0.08198701] | Reward: 53.4 | Obj1: 58.6 | Obj2: -5.2
Episode 200/500 | Steps: 200 | Weight: [0.95670668 0.04329332] | Reward: 59.9 | Obj1: 62.9 | Obj2: -5.7
Episode 300/500 | Steps: 200 | Weight: [0.57550636 0.42449364] | Reward: 31.9 | Obj1: 59.3 | Obj2: -5.2
Episode 400/500 | Steps: 200 | Weight: [0.12488584 0.87511416] | Reward: 2.2 | Obj1: 61.9 | Obj2: -6.3
Episode 500/500 | Steps: 200 | Weight: [0.01914355 0.98085645] | Reward: -4.4 | Obj1: 64.0 | Obj2: -5.8
训练完成时间: 2025-03-26 03:39:16
训练完成,耗时: 120.95秒

七、总结与扩展

本文实现了多目标强化学习的核心范式——基于动态权重的标量化方法,展示了帕累托前沿的探索能力。

在下一篇文章中,我们将探索 稳定扩散模型(Stable Diffusion),并实现文本到图像生成(Text-to-Image Generation)的完整流程!


注意事项

1.安装依赖:

pip install gymnasium torch numpy

2.自定义环境需继承 gym.Env 并实现 reset()step() 方法。

3.动态权重调整策略可根据实际需求设计(如基于任务难度或用户偏好)。

相关推荐

Python 操作excel的坑__真实的行和列

大佬给的建议__如何快速处理excelopenpyxl库操作excel的时候,单个表的数据量大一些处理速度还能接受,如果涉及多个表甚至多个excel文件的时候速度会很慢,还是建议用pandas来处理,...

Python os.path模块使用指南:轻松处理文件路径

前言在Python编程中,文件和目录的操作是非常重要的一部分。为了方便用户进行文件和目录的操作,Python标准库提供了os模块。其中,os.path子模块提供了一些处理文件路径的函数和方法。本文主要...

Python常用内置模块介绍——文件与系统操作详解

Python提供了多个强大的内置模块用于文件和系统操作,下面我将详细介绍最常用的几个模块及其核心功能。1.os模块-操作系统交互...

Python Flask 建站框架实操教程(flask框架网页)

下面我将带您从零开始构建一个完整的Flask网站,包含用户认证、数据库操作和前端模板等核心功能。##第一部分:基础项目搭建###1.创建项目环境```bash...

为你的python程序上锁:软件序列号生成器

序列号很多同学可能开发了非常多的程序了,并且进行了...

PO设计模式全攻略,在 UI 自动化中的实践总结(以企业微信为例)

一、什么是PO设计模式?PO(PageObject)设计模式将某个页面的所有元素对象定位和对元素对象的操作封装成一个Page类,即一个py文件,并以页面为单位来写测试用例,实现页面对象和测试用例的...

这种小工具居然也能在某鱼卖钱?我用Python一天能写...

前两天在某鱼闲逛,本来想找个二手机械键盘,结果刷着刷着突然看到有人在卖——Word批量转PDF小工具...

python打包成exe,程序有图标,但是任务栏和窗口都没有显示图标

代码中指定图标信息#设置应用ID,确保任务栏图标正确显示ifsys.platform=="win32":importctypesapp_id=...

使用Python构建电影推荐系统(用python做推荐系统)

在日常数据挖掘工作中,除了会涉及到使用Python处理分类或预测任务,有时候还会涉及推荐系统相关任务。...

python爬取并分析淘宝商品信息(python爬取淘宝商品数据)

python爬取并分析淘宝商品信息背景介绍一、模拟登陆二、爬取商品信息1.定义相关参数2.分析并定义正则3.数据爬取三、简单数据分析1.导入库2.中文显示3.读取数据4.分析价格分布5.分析销售...

OpenCV入门学习基础教程(从小白变大神)

Opencv是用于快速处理图像处理、计算机视觉问题的工具,支持多种语言进行开发如c++、python、java等,下面这篇文章主要给大家介绍了关于openCV入门学习基础教程的相关资料,需要的朋友可以...

python图像处理-一行代码实现灰度图抠图

抠图是ps的最基本技能,利用python可以实现用一行代码实现灰度图抠图。基础算法是...

从头开始学python:如何用Matplotlib绘图表

Matplotlib是一个用于绘制图表的库。如果你有用过python处理数据,那Matplotlib可以更直观的帮你把数据展示出来。直接上代码看例子:importmatplotlib.pyplot...

Python爬取爱奇艺腾讯视频 250,000 条数据分析为什么李诞不值得了

在《Python爬取爱奇艺52432条数据分析谁才是《奇葩说》的焦点人物?》这篇文章中,我们从爱奇艺爬取了5万多条评论数据,并对一些关键数据进行了分析,由此总结出了一些明面上看不到的数据,并...

Python Matplotlib 库使用基本指南

简介Matplotlib是一个广泛使用的Python数据可视化库,它可以创建各种类型的图表、图形和可视化效果。无论是简单的折线图还是复杂的热力图,Matplotlib提供了丰富的功能来满足我们...

取消回复欢迎 发表评论: