【强化学习】演员评论家Actor-Critic算法(万字长文、附代码)

【强化学习】演员评论家Actor-Critic算法(万字长文、附代码)
        📢本篇文章是博主强化学习(RL)领域学习时,用于个人学习、研究或者欣赏使用,并基于博主对相关等领域的一些理解而记录的学习摘录和笔记,若有不当和侵权之处,指出后将会立即改正,还望谅解。文章分类在👉强化学习专栏:

       【强化学习】- 【单智能体强化学习】(7)---《演员评论家Actor-Critic算法》

演员评论家Actor-Critic算法

目录

Actor-Critic算法理解

1. 角色设定

2. 两者如何协作

3. 学习的核心

4. 为什么叫Actor-Critic?

生活中例子:

Actor-Critic算法的背景与来源

1. 强化学习的起源

2. 策略梯度方法的局限性

3. Actor-Critic的提出

4. 历史发展与应用

Actor-Critic算法流程的推导

1. 强化学习的优化目标

2. 策略梯度定理

3. Critic:值函数估计

4. Actor:策略优化

5. 完整算法流程

[Python] Actor-Critic算法实现

算法伪代码

算法示例代码

Actor-Critic算法实战代码

算法测试代码

[Notice]  关键点总结

总结


Actor-Critic算法理解

        Actor-Critic算法是一种强化学习中的方法,结合了“演员”(Actor)和“评论家”(Critic)两个部分。下面用一个生活中的比喻来说明它的原理:

1. 角色设定

想象你是一名学习爬山的机器人,而你的目标是找到山顶(获得最高的奖励)。在爬山过程中:

  • Actor(行动者):它就像一个“冒险家”,负责决定下一步往哪里走(比如往左一步还是往右一步)。但它并不总是很聪明,可能会选错方向。
  • Critic(评论者):它就像一个“导师”,站在一旁,评价冒险家的表现。它会告诉Actor:“这一步走得好,接近山顶了”或者“走错了,离山顶更远了”。

2. 两者如何协作

Actor-Critic算法的运作过程大致如下:

  • **Actor(冒险家)**观察环境(如坡度、方向),根据它的“策略”(Policy)选择一个动作(比如往左走)。
  • **Critic(导师)**会根据冒险家的动作和环境的反馈(如高度增加或减少),计算一个“价值”(Value),来表示这个动作的好坏。
  • Actor根据Critic的评价,调整自己的策略,使未来能更聪明地选择动作。

3. 学习的核心

  • Actor的目标:学习一个好的策略,尽可能选择能达到山顶的动作。
  • Critic的目标:准确地评估每一步的表现,帮助Actor改进。

通过这种合作方式,Actor不断优化动作策略,而Critic不断提升评价的准确性。

4. 为什么叫Actor-Critic?

这个名字直接反映了两者的分工:

  • Actor负责行动(选择动作)。
  • Critic负责评价(估算价值)。

两者的结合比单独使用Actor或Critic效果更好,因为它们互相弥补了对方的不足。

生活中例子:

        就像你学习开车,你是Actor,根据道路选择要踩油门还是刹车,而你的驾驶教练就是Critic,告诉你哪个动作更安全、更接近目标。


Actor-Critic算法的背景与来源

        Actor-Critic算法是强化学习领域的一种重要方法,它结合了值函数估计策略优化的优点。在理解其背景时,需要从强化学习的演化历史、策略梯度方法的局限性以及如何通过值函数辅助优化策略展开。

1. 强化学习的起源

        强化学习的目标是使智能体通过与环境的交互,学会在不同状态下选择最优动作,从而最大化长期收益。主要研究方法可以分为以下几类:

  1. 值函数方法(如Q学习):估算每个状态或状态-动作对的价值,并依据最大价值选择动作;
  2. 策略方法:直接优化动作选择的概率分布(策略),通过采样环境反馈进行改进;
  3. 策略-值函数结合的方法:例如Actor-Critic,综合两者的优点。

        随着强化学习问题复杂度的增加,仅依赖值函数方法会面临高维状态空间下的维度灾难,而纯策略方法在优化过程中可能收敛速度较慢。因此,结合策略与值函数的Actor-Critic应运而生。

2. 策略梯度方法的局限性

        策略梯度方法通过优化策略函数直接解决强化学习问题,核心思想是通过以下公式更新策略参数

\theta

\nabla_\theta J(\theta) = \mathbb{E}{\pi\theta} \left[ \nabla_\theta \log \pi_\theta(a|s) \cdot A^\pi(s, a) \right]

其中

A^\pi(s, a)

是优势函数,用于衡量动作的相对好坏。

局限性:
  1. 高方差:直接使用环境反馈(奖励)计算梯度会导致策略梯度的方差很高,影响优化效率;
  2. 低效率:由于奖励信号传递较慢,可能需要大量采样才能学到有效的策略。

为了解决这些问题,研究者引入了Critic,用于降低方差并加速策略优化。

3. Actor-Critic的提出

3.1 概念来源

Actor-Critic算法由策略梯度值函数估计结合而成:

  • Actor(行动者):策略网络,决定在每个状态下采取的动作;
  • Critic(评论者):值函数网络,估算当前状态或状态-动作对的价值,用于指导Actor改进。

这一框架的核心思想是利用Critic降低策略梯度的方差,同时保留策略方法的灵活性。

3.2 数学依据

        Critic通过估算值函数

V^\pi(s)

Q^\pi(s, a)

来计算时间差分(TD)误差

\delta_t = r_t + \gamma V^\pi(s_{t+1}) - V^\pi(s_t)
  • Critic最小化TD误差的平方,学习状态值函数;
  • Actor利用TD误差调整策略,使得策略向更优的方向发展。

这一机制使Actor-Critic算法既可以高效地采样环境反馈,又能够快速调整策略参数。

4. 历史发展与应用

4.1 最早提出

        Actor-Critic算法最早由Sutton等人提出(1980年代),作为策略梯度方法的变体,用于解决高方差问题。

4.2 演化与扩展
  1. A3C(Asynchronous Advantage Actor-Critic)
    • 提出时间:2016年,由DeepMind引入。
    • 关键点:通过多线程并行化显著提升学习效率。
  2. PPO(Proximal Policy Optimization)
    • 提出时间:2017年,由OpenAI提出。
    • 关键点:限制策略更新的幅度,改进稳定性。

Actor-Critic算法流程的推导

        Actor-Critic算法结合了策略梯度方法(Policy Gradient)和值函数估计,核心是通过Actor(策略函数)选择动作,通过Critic(值函数)评估这些动作,并相互协作改进。以下是基于数学公式推导的算法流程。

1. 强化学习的优化目标

        目标是最大化累积折扣奖励的期望

J(\theta) = \mathbb{E}{\pi\theta} \left[ \sum_{t=0}^\infty \gamma^t r_t \right] ]

其中:

V^\pi(s)
\gamma

:折扣因子,控制未来奖励的权重。

r_t

:时间  t  的即时奖励;

\pi_\theta(a|s)

:策略函数,表示在状态  s  下选择动作  a  的概率;

2. 策略梯度定理

        为了优化策略函数

\pi_\theta

,我们计算目标函数

J(\theta)

对参数

\theta

的梯度:

\nabla_\theta J(\theta) = \mathbb{E}{\pi\theta} \left[ \nabla_\theta \log \pi_\theta(a|s) \cdot A^\pi(s, a) \right]
A^\pi(s, a)

:优势函数,衡量动作

a

的相对优势。

\nabla_\theta \log \pi_\theta(a|s)

:策略的对数梯度,指示如何调整策略参数以提升选取当前动作的概率;

优势函数的估计:

A^\pi(s, a) \approx r + \gamma V^\pi(s') - V^\pi(s)

其中:

s'

:动作

a

执行后的下一状态。

V^\pi(s)

:状态值函数,表示在状态

s

时累积奖励的期望;

3. Critic:值函数估计

                Critic的目标是通过最小化均方误差,学习状态值函数

V^\pi(s)

L(w) = \frac{1}{2} \mathbb{E} \left[ \left( r + \gamma V^\pi(s') - V^\pi(s) \right)^2 \right]
V^\pi(s)

通常由神经网络近似。

参数

w

是Critic网络的权重;

Critic的梯度更新公式:

\nabla_w L(w) = \left( r + \gamma V^\pi(s') - V^\pi(s) \right) \nabla_w V^\pi(s)

4. Actor:策略优化

        Actor根据Critic的反馈来优化策略参数

\theta

。更新公式为:

\theta \leftarrow \theta + \alpha \cdot \nabla_\theta \log \pi_\theta(a|s) \cdot \delta

其中:

\alpha

:学习率。

\delta = r + \gamma V^\pi(s') - V^\pi(s)

:时间差分(TD)误差,衡量当前状态值预测的偏差;

Actor的更新方向由Critic计算的TD误差指导。

5. 完整算法流程

结合上述部分,Actor-Critic的算法流程如下:

  1. 重复以下步骤直到收敛:
    • 执行动作  a ,获得奖励  r 和下一状态  s' ;

Actor更新:

\theta \leftarrow \theta + \alpha \cdot \nabla_\theta \log \pi_\theta(a|s) \cdot \delta

Critic更新:

w \leftarrow w + \beta \cdot \delta \cdot \nabla_w V^\pi(s)

Critic计算TD误差:

\delta = r + \gamma V^\pi(s') - V^\pi(s)

在状态  s  下,Actor根据

\pi_\theta(a|s)

采样动作  a ;

初始化Actor和Critic网络的参数

\theta, w


[Python] Actor-Critic算法实现

算法伪代码

        结合上述公式,以下是Actor-Critic的简化伪代码:

# 初始化Actor和Critic的参数 theta = 初始化Actor参数 w = 初始化Critic参数 for episode in range(最大迭代次数): 初始化环境 s = 初始状态 while not done: # Actor选择动作 a = 从π_theta(s)中采样动作 # 执行动作并获得奖励和下一状态 s_next, r, done = 环境.step(a) # Critic评估当前状态 V_s = Critic网络预测值(s, w) V_s_next = Critic网络预测值(s_next, w) # 计算TD误差 delta = r + gamma * V_s_next - V_s # 更新Critic参数 w = w + alpha_critic * delta * ∇_w V_s # 更新Actor参数 theta = theta + alpha_actor * delta * ∇_theta log π_theta(a | s) # 更新状态 s = s_next 

算法示例代码

以下是使用PyTorch实现的Actor-Critic算法的示例代码:

import numpy as np import torch import torch.nn as nn import torch.optim as optim # Actor网络 class Actor(nn.Module): def __init__(self, state_dim, action_dim): super(Actor, self).__init__() self.fc = nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, action_dim), nn.Softmax(dim=-1) ) def forward(self, state): return self.fc(state) # Critic网络 class Critic(nn.Module): def __init__(self, state_dim): super(Critic, self).__init__() self.fc = nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 1) ) def forward(self, state): return self.fc(state) # Actor-Critic算法 class ActorCritic: def __init__(self, state_dim, action_dim, gamma=0.99, lr=1e-3): self.actor = Actor(state_dim, action_dim) self.critic = Critic(state_dim) self.gamma = gamma self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=lr) self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=lr) def select_action(self, state): state = torch.tensor(state, dtype=torch.float32) probs = self.actor(state) action = torch.multinomial(probs, 1).item() return action, probs[action] def update(self, state, action_prob, reward, next_state, done): state = torch.tensor(state, dtype=torch.float32) next_state = torch.tensor(next_state, dtype=torch.float32) reward = torch.tensor(reward, dtype=torch.float32) done = torch.tensor(done, dtype=torch.float32) # Critic更新 value = self.critic(state) next_value = self.critic(next_state) target = reward + self.gamma * next_value * (1 - done) td_error = target - value critic_loss = td_error.pow(2) self.critic_optimizer.zero_grad() critic_loss.backward() self.critic_optimizer.step() # Actor更新 actor_loss = -torch.log(action_prob) * td_error.detach() self.actor_optimizer.zero_grad() actor_loss.backward() self.actor_optimizer.step() 
 项目代码我已经放入GitCode里面,可以通过下面链接跳转:🔥

【强化学习】--- 演员评论家Actor-Critic算法 

后续相关单智能体强化学习算法也会不断在【强化学习】项目里更新,如果该项目对你有所帮助,请帮我点一个星星✨✨✨✨✨,鼓励分享,十分感谢!!!

若是下面代码复现困难或者有问题,也欢迎评论区留言

Actor-Critic算法实战代码

下面是基于Python和PyTorch的Actor-Critic算法的项目实代码:

Actor-->Policy网络

"""《Actor-Critic算法》 时间:2024.12 作者:不去幼儿园 """ import torch from torch import nn from torch.nn import functional as F import numpy as np # ------------------------------------ # # 策略梯度Actor,动作选择 # ------------------------------------ # class PolicyNet(nn.Module): def __init__(self, n_states, n_hiddens, n_actions): super(PolicyNet, self).__init__() self.fc1 = nn.Linear(n_states, n_hiddens) self.fc2 = nn.Linear(n_hiddens, n_actions) # 前向传播 def forward(self, x): x = self.fc1(x) # [b,n_states]-->[b,n_hiddens] x = F.relu(x) x = self.fc2(x) # [b,n_hiddens]-->[b,n_actions] # 每个状态对应的动作的概率 x = F.softmax(x, dim=1) # [b,n_actions]-->[b,n_actions] return x

Critic-->Value网络

# ------------------------------------ # # 值函数Critic,动作评估输出 shape=[b,1] # ------------------------------------ # class ValueNet(nn.Module): def __init__(self, n_states, n_hiddens): super(ValueNet, self).__init__() self.fc1 = nn.Linear(n_states, n_hiddens) self.fc2 = nn.Linear(n_hiddens, 1) # 前向传播 def forward(self, x): x = self.fc1(x) # [b,n_states]-->[b,n_hiddens] x = F.relu(x) x = self.fc2(x) # [b,n_hiddens]-->[b,1] return x 
Actor-Critic算法
# ------------------------------------ # # Actor-Critic # ------------------------------------ # class ActorCritic: def __init__(self, n_states, n_hiddens, n_actions, actor_lr, critic_lr, gamma): # 属性分配 self.gamma = gamma # 实例化策略网络 self.actor = PolicyNet(n_states, n_hiddens, n_actions) # 实例化价值网络 self.critic = ValueNet(n_states, n_hiddens) # 策略网络的优化器 self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr) # 价值网络的优化器 self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr) # 动作选择 def take_action(self, state): # 维度变换numpy[n_states]-->[1,n_sates]-->tensor state = torch.tensor(state[np.newaxis, :]) # 动作价值函数,当前状态下各个动作的概率 probs = self.actor(state) # 创建以probs为标准类型的数据分布 action_dist = torch.distributions.Categorical(probs) # 随机选择一个动作 tensor-->int action = action_dist.sample().item() return action # 模型更新 def update(self, transition_dict): # 训练集 states = torch.tensor(transition_dict['states'], dtype=torch.float) actions = torch.tensor(transition_dict['actions']).view(-1,1) rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1,1) next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float) dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1,1) # 预测的当前时刻的state_value td_value = self.critic(states) # 目标的当前时刻的state_value td_target = rewards + self.gamma * self.critic(next_states) * (1-dones) # 时序差分的误差计算,目标的state_value与预测的state_value之差 td_delta = td_target - td_value # 对每个状态对应的动作价值用log函数 log_probs = torch.log(self.actor(states).gather(1, actions)) # 策略梯度损失 actor_loss = torch.mean(-log_probs * td_delta.detach()) # 值函数损失,预测值和目标值之间 critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach())) # 优化器梯度清0 self.actor_optimizer.zero_grad() # 策略梯度网络的优化器 self.critic_optimizer.zero_grad() # 价值网络的优化器 # 反向传播 actor_loss.backward() critic_loss.backward() # 参数更新 self.actor_optimizer.step() self.critic_optimizer.step() 

算法测试代码

有一个简单的CartPole环境,以下是训练代码:

import numpy as np import matplotlib.pyplot as plt import gym import torch from Actor_Critic import ActorCritic # ----------------------------------------- # # 参数设置 # ----------------------------------------- # num_episodes = 100 # 总迭代次数 gamma = 0.9 # 折扣因子 actor_lr = 1e-3 # 策略网络的学习率 critic_lr = 1e-2 # 价值网络的学习率 n_hiddens = 16 # 隐含层神经元个数 env_name = 'CartPole-v1' return_list = [] # 保存每个回合的return # ----------------------------------------- # # 环境加载 # ----------------------------------------- # env = gym.make(env_name, render_mode="human") n_states = env.observation_space.shape[0] # 状态数 4 n_actions = env.action_space.n # 动作数 2 # ----------------------------------------- # # 模型构建 # ----------------------------------------- # agent = ActorCritic(n_states=n_states, # 状态数 n_hiddens=n_hiddens, # 隐含层数 n_actions=n_actions, # 动作数 actor_lr=actor_lr, # 策略网络学习率 critic_lr=critic_lr, # 价值网络学习率 gamma=gamma) # 折扣因子 # ----------------------------------------- # # 训练--回合更新 # ----------------------------------------- # for i in range(num_episodes): state = env.reset()[0] # 环境重置 done = False # 任务完成的标记 episode_return = 0 # 累计每回合的reward # 构造数据集,保存每个回合的状态数据 transition_dict = { 'states': [], 'actions': [], 'next_states': [], 'rewards': [], 'dones': [], } while not done: action = agent.take_action(state) # 动作选择 next_state, reward, done, _, _ = env.step(action) # 环境更新 # 保存每个时刻的状态\动作\... transition_dict['states'].append(state) transition_dict['actions'].append(action) transition_dict['next_states'].append(next_state) transition_dict['rewards'].append(reward) transition_dict['dones'].append(done) # 更新状态 state = next_state # 累计回合奖励 episode_return += reward # 保存每个回合的return return_list.append(episode_return) # 模型训练 agent.update(transition_dict) # 打印回合信息 print(f'iter:{i}, return:{np.mean(return_list[-10:])}') # -------------------------------------- # # 绘图 # -------------------------------------- # plt.plot(return_list) plt.title('return') plt.show() 

[Notice]  关键点总结

  1. Critic的稳定性:Critic的误差直接影响Actor的梯度更新。
  2. 熵正则化:为了鼓励探索,可以对Actor的损失函数加入熵项。
  3. 多线程优化:使用A3C(Asynchronous Advantage Actor-Critic)可以提升性能。
  4. PPO改进:限制更新范围,解决策略更新过程中的不稳定性。
​# 环境配置 Python 3.11.5 torch 2.1.0 torchvision 0.16.0 gym 0.26.2

总结

        Actor-Critic算法的提出源于策略梯度方法的高方差问题,通过结合值函数(Critic)降低优化方差,提高学习效率。随着强化学习的不断发展,Actor-Critic及其扩展(如A3C、PPO)成为复杂任务中广泛使用的算法。

 更多强化学习文章,请前往:【强化学习(RL)】专栏

        博客都是给自己看的笔记,如有误导深表抱歉。文章若有不当和不正确之处,还望理解与指出。由于部分文字、图片等来源于互联网,无法核实真实出处,如涉及相关争议,请联系博主删除。如有错误、疑问和侵权,欢迎评论留言联系作者,或者添加VX:Rainbook_2,联系作者。✨

Read more

AgentScope Java多智能体框架

1. 技术架构与功能介绍 AgentScope Java 的核心设计理念是 “Agent-Oriented Programming” (面向智能体编程)。 核心功能 * ReAct 范式驱动:内置推理-行动(Reasoning-Acting)循环,智能体能自主规划步骤并调用工具。 * 响应式内核:基于 Project Reactor (Mono/Flux),天然支持非阻塞 I/O,适合处理高并发的 Agent 请求。 * 人类在环 (HITL):支持随时暂停 Agent 执行,接入人工干预后再恢复,这在企业级应用中至关重要。 * 多协议集成:支持 MCP (Model Context Protocol) 协议,可以无缝调用外部各种工具服务。 架构图示 源码级组件解析 从源码结构看,agentscope-java 主要由以下四大基石组成: 1. Msg (消息对象)

By Ne0inhk
Elasticsearch核心概念与Java客户端实战 构建高性能搜索服务

Elasticsearch核心概念与Java客户端实战 构建高性能搜索服务

目录 🎯 先说说我被ES"虐惨"的经历 ✨ 摘要 1. 为什么选择Elasticsearch? 1.1 从数据库的痛苦说起 1.2 Elasticsearch的优势 2. ES核心架构解析 2.1 集群架构 2.2 索引与分片 3. Java客户端实战 3.1 客户端选型对比 3.2 RestHighLevelClient配置 3.3 Spring Data Elasticsearch配置 4. 索引设计最佳实践 4.1 索引生命周期管理 4.2 映射设计技巧 5. 查询优化实战 5.1 查询类型对比 5.

By Ne0inhk
【Java 开发日记】我们来说一下消息的可靠性投递

【Java 开发日记】我们来说一下消息的可靠性投递

目录 1. 核心概念 2. 面临的挑战 3. 关键实现机制 3.1 生产端保证 3.2 Broker端保证 3.3 消费端保证 4. 完整可靠性方案 4.1 事务消息方案(如RocketMQ) 4.2 最大努力投递方案 4.3 本地消息表方案(经典) 5. 高级特性与优化 5.1 顺序性保证 5.2 批量消息可靠性 5.3 监控与对账 6. 不同MQ的实现差异 7. 实践建议 总结 面试回答 1. 核心概念 可靠性投递(Reliable

By Ne0inhk
JAVA多线程并发编程:并发容器与线程协作实战

JAVA多线程并发编程:并发容器与线程协作实战

JAVA多线程并发编程:并发容器与线程协作实战 💡 学习目标:掌握JAVA中常用并发容器的特性与适用场景,理解线程间协作的核心原理,能够运用并发容器和协作工具解决实际并发问题。 💡 学习重点:并发容器与普通容器的区别、ConcurrentHashMap 核心原理、CountDownLatch/CyclicBarrier/Semaphore 的使用、生产者消费者模式实现。 1.1 为什么需要并发容器? 在多线程场景下,普通的集合容器(如 HashMap、ArrayList)是线程不安全的。多个线程同时对其进行读写操作时,会导致数据错乱、ConcurrentModificationException 异常等问题。 ⚠️ 注意事项:即使使用 Collections.synchronizedXXX() 方法包装普通容器,也只是通过 synchronized 实现简单的加锁。这种方式锁粒度较粗,并发性能较低。 ✅ 核心结论:并发容器是JAVA为多线程场景设计的高性能容器。它们通过细粒度锁或无锁算法实现线程安全,能够在保证数据一致性的同时,大幅提升并发访问效率。 1.2 常用并

By Ne0inhk