【MADRL】多智能体双延迟深度确定性策略梯度（MATD3）算法

Ne0inhk

— 13 min read
【MADRL】多智能体双延迟深度确定性策略梯度（MATD3）算法
        本篇文章是博主强化学习RL领域学习时，用于个人学习、研究或者欣赏使用，并基于博主对相关等领域的一些理解而记录的学习摘录和笔记，若有不当和侵权之处，指出后将会立即改正，还望谅解。文章分类在强化学习专栏：

       强化学习（7）---《【MADRL】多智能体双延迟深度确定性策略梯度（MATD3）算法》

多智能体双延迟深度确定性策略梯度（MATD3）算法

目录

1.MATD3算法介绍

2.背景

3.算法结构

4.具体公式

5.算法流程

6.公式总结

7.优势与应用场景

8.总结

 [Python] MATD3实现（可移植）

1.MATD3算法介绍

        MATD3（Multi-Agent Twin Delayed Deep Deterministic Policy Gradient）是基于TD3（Twin Delayed DDPG）算法的多智能体版本。TD3是深度确定性策略梯度（DDPG）算法的一个改进版本，主要针对其在确定性策略学习中的一些不稳定性进行了增强。MATD3则扩展了TD3，使其能够在多智能体环境下进行训练和执行。

文章：Addressing Function Approximation Error in Actor-Critic Methods

代码：MADRL多智能体双延迟深度确定性策略梯度（MATD3）算法

 其他多智能体深度强化学习（MADRL）算法见下面博客：

【MADRL】多智能体深度强化学习《纲要》

2.背景

        DDPG算法用于连续动作空间的强化学习任务，但在复杂环境下容易出现策略估计偏差、探索不足等问题。TD3通过引入两种关键机制来解决这些问题：

  • 延迟更新：Actor的更新频率较Critic更低，以避免策略更新过快导致的不稳定性。
  • 目标策略平滑：在计算目标Q值时，给动作增加噪声以减小过估计偏差。

        在多智能体场景中，每个智能体不仅要与环境交互，还需要适应其他智能体的行为。MATD3结合了TD3的稳定性增强机制，并将其应用到多智能体系统中，使其能够在混合协作与竞争的环境下表现更佳。

3.算法结构

        MATD3算法同样是基于Actor-Critic架构的多智能体强化学习算法，其中每个智能体都有独立的Actor网络和双Critic网络。该算法采用集中式训练，分布式执行的结构：

  • 集中式训练：训练过程中，每个智能体的Critic网络可以访问所有智能体的状态和动作，以最大化每个智能体的累积回报。
  • 分布式执行：在执行阶段，智能体仅根据自身的观测来选择动作。

4.具体公式

  1. 环境设定：

延迟更新：为了进一步提高稳定性，Actor网络的更新频率低于Critic网络。例如，Critic网络每更新两次，Actor网络更新一次。此外，目标网络的参数更新也较慢，遵循“软更新”策略：

[ \theta'_i \leftarrow \tau \theta_i + (1 - \tau) \theta'_i ]

其中

( \tau )

是软更新的速率参数。

目标策略平滑：为了减少估计偏差，在计算目标Q值时，给每个智能体的动作

( a'_i )

加入噪声：

[ a'_i = \pi'_i(o'_i) + \epsilon, \quad \epsilon \sim \mathcal{N}(0, \sigma) ]

这样可以避免策略过拟合某些特定的动作，从而提高策略的鲁棒性。

Actor网络：Actor网络的更新也是通过最大化Critic网络的Q值来进行的。Actor策略的梯度可以通过下式计算：

[ \nabla_{\theta_{\pi_i}} J(\pi_i) = \mathbb{E}{s,a} \left[ \nabla{a_i} Q_i^{1}(s, a_1, ..., a_N) \nabla_{\theta_{\pi_i}} \pi_i(o_i) \right] ]

其中只使用

( Q_i^{1} )

来更新Actor策略。

使用均方误差（MSE）损失函数更新两个Critic网络的参数

( \theta_i^{1}, \theta_i^{2} )

：

[ L(\theta_i^{j}) = \mathbb{E}_{s,a,r,s'} \left[ \left( Q_i^{j}(s, a_1, ..., a_N; \theta_i^{j}) - y_i \right)^2 \right] ]

其中

( j = 1, 2 )

。

其中

( Q*_i^{1}, Q*_i^{2} )

是目标网络，

( a'_j )

是通过目标Actor网络

( \pi'_j )

生成的动作。

首先计算目标Q值，使用所有智能体的下一个动作

( a'j )

：

[ y_i = r_i + \gamma \min{j=1,2} Q*_i^{j}(s', a'_1, ..., a'_N) ]

Critic网络：MATD3每个智能体

( i )

的Critic网络有两个Q值函数

( Q_i^{1}(s, a_1, ..., a_N) )

和 

(Q_i^{2}(s, a_1, ..., a_N) )

用于减少Q值估计的偏差。这两个Q值函数的更新方式类似于TD3中的方式：

目标：每个智能体

( i )

的目标是最大化其期望累积回报：

[ R_i = \mathbb{E} \left[ \sum_{t=0}^{T} \gamma^t r_i^t \right] ]

其中，

( r_i^t )

是智能体

( i )

在时刻

( t )

的即时奖励，

( \gamma )

是折扣因子。

每个智能体根据其策略

( \pi_i(o_i) )

选择动作

( a_i )

，并根据奖励函数

( r_i )

得到即时奖励。

系统状态为

( s \in S )

，每个智能体

( i )

的观测为

( o_i \in O )

，动作为

( a_i \in A )

。

5.算法流程

  1. 更新Critic网络：
    • 从经验回放池中采样一个批次数据。
  2. 延迟更新Actor网络：
    • 每隔若干步，更新Actor网络的参数，最大化其对应的Critic网络的Q值。
  3. 软更新目标网络：更新目标Critic网络和目标Actor网络的参数。
  4. 重复步骤，直到智能体在环境中学会优化策略。

根据上述公式计算目标Q值

( y_i )

，并更新两个Critic网络。

交互与经验存储：每个智能体

( i )

与环境交互，记录当前状态、动作、奖励和下一个状态。

初始化：为每个智能体初始化两个Critic网络

( Q_i^{1}, Q_i^{2} )

和一个Actor网络

( \pi_i )

，并初始化对应的目标网络

( Q*_i^{1}, Q*_i^{2}, \pi'_i )

。

6.公式总结

Actor更新：

[ \nabla_{\theta_{\pi_i}} J(\pi_i) = \mathbb{E}{s,a} \left[ \nabla{a_i} Q_i^{1}(s, a_1, ..., a_N) \nabla_{\theta_{\pi_i}} \pi_i(o_i) \right] ]

Critic更新：

[ L(\theta_i^{j}) = \mathbb{E}{s,a,r,s'} \left[ \left( Q_i^{j}(s, a_1, ..., a_N; \theta_i^{j}) - \left( r_i + \gamma \min{j=1,2} Q*_i^{j}(s', a'_1, ..., a'_N) \right) \right)^2 \right] ]

7.优势与应用场景

  • 减少Q值估计偏差：通过引入两个Critic网络，MATD3显著减少了单个Critic在更新过程中的过估计问题，从而提高了稳定性。
  • 解决多智能体非平稳性问题：多智能体环境下，其他智能体的策略会影响每个智能体的策略学习。MATD3通过全局信息的中心化训练方式，使得每个智能体能够学习到更加鲁棒的策略。
  • 混合协作和竞争环境：该算法特别适用于协作与竞争混合的环境，因为它能够处理多个智能体之间的复杂交互。

8.总结

        MATD3算法是TD3算法在多智能体场景下的扩展，通过中心化的Critic结构和去中心化的Actor结构，MATD3能够有效应对多智能体环境下的挑战。算法通过双Critic结构减少Q值估计偏差，并且延迟更新机制进一步提高了训练过程的稳定性，使其在混合协作与竞争的复杂环境中具有良好的表现。

 [Python] MATD3实现（可移植）

        若是下面代码复现困难或者有问题，欢迎评论区留言；需要以整个项目形式的代码，请在评论区留下您的邮箱，以便于及时分享给您（私信难以及时回复）。

主文件：MATD3_main 

import torch import numpy as np from torch.utils.tensorboard import SummaryWriter from environment import Env import argparse from replay_buffer import ReplayBuffer # from maddpg import MADDPG from matd3 import MATD3 import copy class Runner: def __init__(self, args, env_name, number, seed): self.args = args self.env_name = env_name self.number = number self.seed = seed # Create env self.env = Env(env_name, discrete=False) # Continuous action space self.env_evaluate = Env(env_name, discrete=False) self.args.N = self.env.n # The number of agents self.args.obs_dim_n = [self.env.observation_space[i].shape[0] for i in range(self.args.N)] # obs dimensions of N agents self.args.action_dim_n = [self.env.action_space[i].shape[0] for i in range(self.args.N)] # actions dimensions of N agents print("observation_space=", self.env.observation_space) print("obs_dim_n={}".format(self.args.obs_dim_n)) print("action_space=", self.env.action_space) print("action_dim_n={}".format(self.args.action_dim_n)) # Set random seed np.random.seed(self.seed) torch.manual_seed(self.seed) # Create N agents if self.args.algorithm == "MADDPG": print("Algorithm: MADDPG") # self.agent_n = [MADDPG(args, agent_id) for agent_id in range(args.N)] elif self.args.algorithm == "MATD3": print("Algorithm: MATD3") self.agent_n = [MATD3(args, agent_id) for agent_id in range(args.N)] else: print("Wrong!!!") self.replay_buffer = ReplayBuffer(self.args) # Create a tensorboard self.writer = SummaryWriter(log_dir='runs/{}/{}_env_{}_number_{}_seed_{}'.format(self.args.algorithm, self.args.algorithm, self.env_name, self.number, self.seed)) self.evaluate_rewards = [] # Record the rewards during the evaluating self.total_steps = 0 self.noise_std = self.args.noise_std_init # Initialize noise_std def run(self, ): self.evaluate_policy() while self.total_steps < self.args.max_train_steps: obs_n = self.env.reset() for _ in range(self.args.episode_limit): # Each agent selects actions based on its own local observations(add noise for exploration) a_n = [agent.choose_action(obs, noise_std=self.noise_std) for agent, obs in zip(self.agent_n, obs_n)] # --------------------------!!!注意！！！这里一定要deepcopy，MPE环境会把a_n乘5------------------------------------------- obs_next_n, r_n, done_n, _ = self.env.step(copy.deepcopy(a_n)) # Store the transition self.replay_buffer.store_transition(obs_n, a_n, r_n, obs_next_n, done_n) obs_n = obs_next_n self.total_steps += 1 # Decay noise_std if self.args.use_noise_decay: self.noise_std = self.noise_std - self.args.noise_std_decay if self.noise_std - self.args.noise_std_decay > self.args.noise_std_min else self.args.noise_std_min if self.replay_buffer.current_size > self.args.batch_size: # Train each agent individually for agent_id in range(self.args.N): self.agent_n[agent_id].train(self.replay_buffer, self.agent_n) if self.total_steps % self.args.evaluate_freq == 0: self.evaluate_policy() if all(done_n): break self.env.close() self.env_evaluate.close() def evaluate_policy(self, ): evaluate_reward = 0 for _ in range(self.args.evaluate_times): obs_n = self.env_evaluate.reset() episode_reward = 0 for _ in range(self.args.episode_limit): a_n = [agent.choose_action(obs, noise_std=0) for agent, obs in zip(self.agent_n, obs_n)] # We do not add noise when evaluating obs_next_n, r_n, done_n, _ = self.env_evaluate.step(copy.deepcopy(a_n)) episode_reward += r_n[0] obs_n = obs_next_n if all(done_n): break evaluate_reward += episode_reward evaluate_reward = evaluate_reward / self.args.evaluate_times self.evaluate_rewards.append(evaluate_reward) print("total_steps:{} \t evaluate_reward:{} \t noise_std:{}".format(self.total_steps, evaluate_reward, self.noise_std)) self.writer.add_scalar('evaluate_step_rewards_{}'.format(self.env_name), evaluate_reward, global_step=self.total_steps) # Save the rewards and models np.save('./data_train/{}_env_{}_number_{}_seed_{}.npy'.format(self.args.algorithm, self.env_name, self.number, self.seed), np.array(self.evaluate_rewards)) for agent_id in range(self.args.N): self.agent_n[agent_id].save_model(self.env_name, self.args.algorithm, self.number, self.total_steps, agent_id) if __name__ == '__main__': parser = argparse.ArgumentParser("Hyperparameters Setting for MADDPG and MATD3 in MPE environment") parser.add_argument("--max_train_steps", type=int, default=int(1e6), help=" Maximum number of training steps") parser.add_argument("--episode_limit", type=int, default=25, help="Maximum number of steps per episode") parser.add_argument("--evaluate_freq", type=float, default=5000, help="Evaluate the policy every 'evaluate_freq' steps") parser.add_argument("--evaluate_times", type=float, default=3, help="Evaluate times") parser.add_argument("--max_action", type=float, default=1.0, help="Max action") parser.add_argument("--algorithm", type=str, default="MATD3", help="MADDPG or MATD3") parser.add_argument("--buffer_size", type=int, default=int(1e6), help="The capacity of the replay buffer") parser.add_argument("--batch_size", type=int, default=1024, help="Batch size") parser.add_argument("--hidden_dim", type=int, default=64, help="The number of neurons in hidden layers of the neural network") parser.add_argument("--noise_std_init", type=float, default=0.2, help="The std of Gaussian noise for exploration") parser.add_argument("--noise_std_min", type=float, default=0.05, help="The std of Gaussian noise for exploration") parser.add_argument("--noise_decay_steps", type=float, default=3e5, help="How many steps before the noise_std decays to the minimum") parser.add_argument("--use_noise_decay", type=bool, default=True, help="Whether to decay the noise_std") parser.add_argument("--lr_a", type=float, default=5e-4, help="Learning rate of actor") parser.add_argument("--lr_c", type=float, default=5e-4, help="Learning rate of critic") parser.add_argument("--gamma", type=float, default=0.95, help="Discount factor") parser.add_argument("--tau", type=float, default=0.01, help="Softly update the target network") parser.add_argument("--use_orthogonal_init", type=bool, default=True, help="Orthogonal initialization") parser.add_argument("--use_grad_clip", type=bool, default=True, help="Gradient clip") # --------------------------------------MATD3-------------------------------------------------------------------- parser.add_argument("--policy_noise", type=float, default=0.2, help="Target policy smoothing") parser.add_argument("--noise_clip", type=float, default=0.5, help="Clip noise") parser.add_argument("--policy_update_freq", type=int, default=2, help="The frequency of policy updates") args = parser.parse_args() args.noise_std_decay = (args.noise_std_init - args.noise_std_min) / args.noise_decay_steps env_names = ["simple_speaker_listener", "simple_spread"] env_index = 0 runner = Runner(args, env_name=env_names[env_index], number=1, seed=0) runner.run()

环境文件：environment

# Please write down your environment Settings # Pay attention to the input and output of parameters class Env: def __init__(self, args, discrete): self.args = args self.discrete = discrete

移植事项：

1.注意环境参数的设置格式

2.注意环境的返回值利用

3.注意主运行流程的runner.run()的相关设置，等

可借鉴：【MADRL】基于MADRL的单调价值函数分解（QMIX）算法​​​​​​ 中关于 QMIX算法移植的注意事项和代码注释。

     文章若有不当和不正确之处，还望理解与指出。由于部分文字、图片等来源于互联网，无法核实真实出处，如涉及相关争议，请联系博主删除。如有错误、疑问和侵权，欢迎评论留言联系作者，或者关注VX公众号：Rain21321，联系作者。

Read more

Spring Cloud Alibaba 详解

Spring Cloud Alibaba 详解

一、核心定位 & 核心价值（面试开篇必答，定调必背） ✅ 什么是 Spring Cloud Alibaba   Spring Cloud Alibaba 是 阿里巴巴开源的一站式微服务解决方案，是 Spring Cloud 生态的核心子项目，完全兼容 Spring Cloud 标准规范，是目前国内企业 95% 以上微服务架构的首选技术栈。 ✅ 诞生背景（面试必考）         原生的 Spring Cloud 核心组件（Eureka、Hystrix、Zuul、Config）基于 Netflix 系列开源组件，但 Netflix 从 2018 年开始对核心组件陆续宣布停更 / 闭源，导致原生 Spring Cloud 存在稳定性、维护性、

By Ne0inhk
【MySQL修炼篇】从S锁/X锁到Next-Key Lock：MySQL锁机制硬核拆解

【MySQL修炼篇】从S锁/X锁到Next-Key Lock：MySQL锁机制硬核拆解

🍃 予枫：个人主页 📚 个人专栏: 《Java 从入门到起飞》《读研码农的干货日常》 💻 Debug 这个世界，Return 更好的自己！ 引言 线上系统突然报出死锁异常，业务数据更新卡住，排查半天却连锁的类型都分不清？行锁、表锁、间隙锁到底有啥区别？S锁和X锁的竞争又是如何引发死锁的？作为后端开发者，数据库锁机制是绕不开的核心知识点，更是保障系统数据一致性和并发性能的关键。本文将从基础锁类型到死锁排查，层层拆解MySQL锁机制，带你吃透每个核心要点，轻松应对线上锁相关问题～ 文章目录 * 引言 * 一、核心锁类型基础：S锁与X锁 * 1.1 共享锁（S锁）：读锁不互斥 * 1.2 排他锁（X锁）：写锁全互斥 * 二、粒度区分：表锁与行锁 * 2.1 表锁：粗粒度锁，高效低并发 * 核心特性： * 适用场景：

By Ne0inhk

Flutter 三方库 functions_framework 的鸿蒙化适配指南 - 掌控云端函数架构、Serverless 微服务实战、鸿蒙级端云一体化专家

欢迎加入开源鸿蒙跨平台社区：https://openharmonycrossplatform.ZEEKLOG.net Flutter 三方库 functions_framework 的鸿蒙化适配指南 - 掌控云端函数架构、Serverless 微服务实战、鸿蒙级端云一体化专家 【百篇巨献：第 100 篇博文里程碑】 在鸿蒙跨平台应用迈向“端云一体化”的征程中，如何快速、低门槛地编写能够运行在各种 Serverless 环境（如 Google Cloud Functions, Knative）的响应函数是每一位架构师的追求。如果你希望在鸿蒙项目中，利用一套极简、符合标准的函数式编程模型来处理 HTTP 请求或 Cloud Events。今天我们要深度解析的 functions_framework——由 Google 维护的标准化 Dart 云函数框架，正是帮你打通“鸿蒙端逻辑”与“

By Ne0inhk
黑马点评完整代码(RabbitMQ优化)+简历编写+面试重点 ⭐

黑马点评完整代码(RabbitMQ优化)+简历编写+面试重点 ⭐

简历上展示黑马点评 完整代码地址 微服务学成在线项目 前言 当初就是当作一个学习笔记和个人面试记录发的，没想到这么多人收藏浏览，还是感慨学Java的人确实多啊。 适合什么人看呢，我仅仅说说我个人的理解，因为我现在也是个经历秋招的双非学生。 1.初学者学习完Redis基础，想来个实战，黑马点评还是特别好的一个项目，基本包含了所有数据类型的运用和redis其他功能的扩展，这篇文章可以带你提炼重点，很好的走下流程。 2.但大部分人是冲着找实习和秋招去的，像我这种学历不高的秋招就不要写黑马点评了，即使包装，也会很容易看出来，我找实习的时候就被面试官问到这是不是黑马点评过，我们可以把其中的闪光点迁移到你找的其他项目中，比如缓存穿透雪崩击穿的解决方法，redisson分布式锁解决一人一单，这种在大多项目中都可以添加，自圆其说就行。 3.对于找实习的像大二，大三上的，想找个小厂试试手垂直向上升的，可以吃透它，面试官问你遇到的困难或者是你觉得难点，就可以重点讲一人一单这个解决方法和流程，越详细越好。 4.前提是大家不用直接用这套模板，太多人用了，这也是我从网上找的别人的，巧用AI让它改改项

By Ne0inhk