当前位置： 首页 > news >正文

强化学习的常用策略浅析

news 来源：原创 2025/5/15 14:24:17

目录

DDPG（深度确定性策略梯度）

DPPO（分布式近端策略优化）

simply_PPO（简单近端策略优化）

discrete_DPPO（离散动作的分布式近端策略优化）

DDPG（深度确定性策略梯度）

算法概述

DDPG 是一种适用于连续动作空间的无模型强化学习算法，结合了深度 Q 网络（DQN）和确定性策略梯度（DPG）的思想。它使用两个神经网络：一个是策略网络（Actor），用于生成动作；另一个是价值网络（Critic），用于评估动作的价值。

代码结构与功能

超参数设置：定义了训练的最大回合数、每回合的最大步数、学习率、折扣因子等。

Actor 类：_build_net 方法：构建策略网络，输出确定性动作。

learn 方法：更新策略网络的参数，并定期更新目标网络。

choose_action 方法：根据当前状态选择动作。

add_grad_to_graph 方法：将 Critic 网络计算的梯度添加到策略网络的更新中。

Critic 类：_build_net 方法：构建价值网络，输出状态 - 动作对的价值。

learn 方法：更新价值网络的参数，并定期更新目标网络。

Memory 类：用于存储和采样经验数据。

训练循环

在每个回合中，智能体与环境交互，收集经验数据并存储到内存中。当内存达到一定容量时，开始从内存中采样数据进行训练。

代码示例

# 初始化环境
env = gym.make(ENV_NAME)
env = env.unwrapped
env.seed(1)

# 定义状态和动作维度
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.shape[0]
action_bound = env.action_space.high

# 创建会话
sess = tf.Session()

# 初始化 Actor 和 Critic 网络
actor = Actor(sess, action_dim, action_bound, LR_A, REPLACE_ITER_A)
critic = Critic(sess, state_dim, action_dim, LR_C, GAMMA, REPLACE_ITER_C, actor.a, actor.a_)

# 将 Critic 计算的梯度添加到 Actor 网络中
actor.add_grad_to_graph(critic.a_grads)

# 初始化全局变量
sess.run(tf.global_variables_initializer())

# 创建经验回放内存
M = Memory(MEMORY_CAPACITY, dims=2 * state_dim + action_dim + 1)

# 训练循环
for i in range(MAX_EPISODES):
    s = env.reset()
    ep_reward = 0
    for j in range(MAX_EP_STEPS):
        if RENDER:
            env.render()
        # 选择动作并添加探索噪声
        a = actor.choose_action(s)
        a = np.clip(np.random.normal(a, var), -2, 2)
        s_, r, done, info = env.step(a)
        # 存储经验数据
        M.store_transition(s, a, r / 10, s_)
        # 开始学习
        if M.pointer > MEMORY_CAPACITY:
            var *= .9995  # 衰减探索噪声
            b_M = M.sample(BATCH_SIZE)
            b_s = b_M[:, :state_dim]
            b_a = b_M[:, state_dim: state_dim + action_dim]
            b_r = b_M[:, -state_dim - 1: -state_dim]
            b_s_ = b_M[:, -state_dim:]
            # 更新 Critic 网络
            critic.learn(b_s, b_a, b_r, b_s_)
            # 更新 Actor 网络
            actor.learn(b_s)
        s = s_
        ep_reward += r
        if j == MAX_EP_STEPS - 1:
            print('Episode:', i, ' Reward: %i' % int(ep_reward), 'Explore: %.2f' % var)
            if ep_reward > -300:
                RENDER = True
            break

DPPO（分布式近端策略优化）

算法概述

DPPO 是一种基于近端策略优化（PPO）的分布式算法，用于处理连续动作空间。它通过多个并行的工作线程收集经验数据，然后由一个全局的 PPO 网络进行更新。

代码结构与功能

超参数设置：定义了训练的最大回合数、每回合的最大步数、学习率、折扣因子等。

PPO 类：_build_anet 方法：构建策略网络，输出动作的概率分布。

update 方法：更新策略网络和价值网络的参数。

choose_action 方法：根据当前状态选择动作。

get_v 方法：获取当前状态的价值。

Worker 类：每个工作线程负责与环境交互，收集经验数据并将其存储到队列中。

训练循环

多个工作线程并行运行，收集经验数据。当队列中的数据达到一定数量时，全局的 PPO 网络从队列中取出数据进行更新。

代码示例

# 初始化全局 PPO 网络
GLOBAL_PPO = PPO()
# 创建事件对象
UPDATE_EVENT, ROLLING_EVENT = threading.Event(), threading.Event()
UPDATE_EVENT.clear()  # 初始时不进行更新
ROLLING_EVENT.set()  # 开始收集数据
# 创建工作线程
workers = [Worker(wid=i) for i in range(N_WORKER)]
# 初始化全局计数器和奖励记录
GLOBAL_UPDATE_COUNTER, GLOBAL_EP = 0, 0
GLOBAL_RUNNING_R = []
# 创建协调器和队列
COORD = tf.train.Coordinator()
QUEUE = queue.Queue()
# 启动工作线程
threads = []
for worker in workers:
    t = threading.Thread(target=worker.work, args=())
    t.start()
    threads.append(t)
# 启动 PPO 更新线程
threads.append(threading.Thread(target=GLOBAL_PPO.update,))
threads[-1].start()
# 等待所有线程结束
COORD.join(threads)
# 绘制奖励变化曲线并进行测试
plt.plot(np.arange(len(GLOBAL_RUNNING_R)), GLOBAL_RUNNING_R)
plt.xlabel('Episode'); plt.ylabel('Moving reward'); plt.ion(); plt.show()
env = gym.make('Pendulum-v1')
while True:
    s = env.reset()
    for t in range(300):
        env.render()
        s = env.step(GLOBAL_PPO.choose_action(s))[0]

simply_PPO（简单近端策略优化）

算法概述

simply_PPO 是一种简化的近端策略优化算法，用于处理连续动作空间。它使用一个策略网络和一个价值网络，通过裁剪代理目标函数来更新策略网络的参数。

代码结构与功能

超参数设置：定义了训练的最大回合数、每回合的最大步数、学习率、折扣因子等。

PPO 类：_build_anet 方法：构建策略网络，输出动作的概率分布。

update 方法：更新策略网络和价值网络的参数。

choose_action 方法：根据当前状态选择动作。

get_v 方法：获取当前状态的价值。

训练循环

在每个回合中，智能体与环境交互，收集经验数据并存储到缓冲区中。当缓冲区中的数据达到一定数量时，开始从缓冲区中采样数据进行训练。

代码示例

# 初始化环境
env = gym.make('Pendulum-v1').unwrapped
# 初始化 PPO 网络
ppo = PPO()
# 记录所有回合的奖励
all_ep_r = []
# 训练循环
for ep in range(EP_MAX):
    s = env.reset()
    buffer_s, buffer_a, buffer_r = [], [], []
    ep_r = 0
    for t in range(EP_LEN):
        env.render()
        # 选择动作
        a = ppo.choose_action(s)
        s_, r, done, _ = env.step(a)
        # 存储经验数据
        buffer_s.append(s)
        buffer_a.append(a)
        buffer_r.append((r + 8) / 8)  # 归一化奖励
        s = s_
        ep_r += r
        # 更新 PPO 网络
        if (t + 1) % BATCH == 0 or t == EP_LEN - 1:
            v_s_ = ppo.get_v(s_)
            discounted_r = []
            for r in buffer_r[::-1]:
                v_s_ = r + GAMMA * v_s_
                discounted_r.append(v_s_)
            discounted_r.reverse()
            bs, ba, br = np.vstack(buffer_s), np.vstack(buffer_a), np.array(discounted_r)[:, np.newaxis]
            buffer_s, buffer_a, buffer_r = [], [], []
            ppo.update(bs, ba, br)
    if ep == 0:
        all_ep_r.append(ep_r)
    else:
        all_ep_r.append(all_ep_r[-1] * 0.9 + ep_r * 0.1)
    print(
        'Ep: %i' % ep,
        "|Ep_r: %i" % ep_r,
        ("|Lam: %.4f" % METHOD['lam']) if METHOD['name'] == 'kl_pen' else '',
    )
# 绘制奖励变化曲线
plt.plot(np.arange(len(all_ep_r)), all_ep_r)
plt.xlabel('Episode'); plt.ylabel('Moving averaged episode reward'); plt.show()

discrete_DPPO（离散动作的分布式近端策略优化）

算法概述

discrete_DPPO 是一种适用于离散动作空间的分布式近端策略优化算法。它通过多个并行的工作线程收集经验数据，然后由一个全局的 PPO 网络进行更新。

代码结构与功能

超参数设置：定义了训练的最大回合数、每回合的最大步数、学习率、折扣因子等。

PPONet 类：_build_anet 方法：构建策略网络，输出动作的概率分布。

update 方法：更新策略网络和价值网络的参数。

choose_action 方法：根据当前状态选择动作。

get_v 方法：获取当前状态的价值。

Worker 类：每个工作线程负责与环境交互，收集经验数据并将其存储到队列中。

训练循环

多个工作线程并行运行，收集经验数据。当队列中的数据达到一定数量时，全局的 PPO 网络从队列中取出数据进行更新。

代码示例

# 初始化全局 PPO 网络
GLOBAL_PPO = PPONet()
# 创建事件对象
UPDATE_EVENT, ROLLING_EVENT = threading.Event(), threading.Event()
UPDATE_EVENT.clear()  # 初始时不进行更新
ROLLING_EVENT.set()  # 开始收集数据
# 创建工作线程
workers = [Worker(wid=i) for i in range(N_WORKER)]
# 初始化全局计数器和奖励记录
GLOBAL_UPDATE_COUNTER, GLOBAL_EP = 0, 0
GLOBAL_RUNNING_R = []
# 创建协调器和队列
COORD = tf.train.Coordinator()
QUEUE = queue.Queue()
# 启动工作线程
threads = []
for worker in workers:
    t = threading.Thread(target=worker.work, args=())
    t.start()
    threads.append(t)
# 启动 PPO 更新线程
threads.append(threading.Thread(target=GLOBAL_PPO.update,))
threads[-1].start()
# 等待所有线程结束
COORD.join(threads)
# 绘制奖励变化曲线并进行测试
plt.plot(np.arange(len(GLOBAL_RUNNING_R)), GLOBAL_RUNNING_R)
plt.xlabel('Episode'); plt.ylabel('Moving reward'); plt.ion(); plt.show()
env = gym.make('CartPole-v0')
while True:
    s = env.reset()
    for t in range(1000):
        env.render()
        s, r, done, info = env.step(GLOBAL_PPO.choose_action(s))
        if done:
            break

相关文章：

  • 【嵌入式学习2】函数
  • 【MySQL】字符集与排序规则
  • unsloth微调QwQ32B(4bit)
  • PyTorch 深度学习实战（22）：多智能体强化学习（MARL）
  • QT5.14.2 gradle 构建安卓失败的处理方法
  • [RoarCTF 2019]Easy Calc-3.23BUUCTF练习day5(2)
  • 大语言模型-2.2/3-主流模型架构与新型架构
  • CH32V208蓝牙内部带运放32位RISC-V工业级微控制器CH32V208CBU6、CH32V208GBU6开发板原理图和PCB
  • 【Linux文件IO】Linux中标准IO的API的描述和基本用法
  • SpringBoot集成Couchbase开发与实践
  • 使用 docker compse 启动 Milvus 修改 MINIO_ACCESS_KEY 导致启动失败
  • onedav一为导航批量自动化导入网址（完整教程）
  • 回归——数学公式推导全过程
  • //TODO 动态代理的本质?
  • 深度学习与计算机视觉方向
  • MATLAB实现基于“蚁群算法”的AMR路径规划
  • 蓝桥杯 回文数组
  • 常见中间件漏洞之四：Apache
  • 通信基本概念
  • windows切换系统版本
  • 证券日报：降准今日正式落地，年内或还有降准空间
  • 陕西省安康市汉阴县县长陈永乐已任汉阴县委书记
  • 4月企业新发放贷款利率处于历史低位
  • “大鼻子情圣”德帕迪约因性侵被判缓刑，还有新的官司等着他
  • 最高降九成！特朗普签署降药价行政令落地存疑，多家跨国药企股价收涨
  • 郑州通报涉“健康证”办理有关问题查处进展情况