OpenAI Gym 中级教程——多智能体系统

Python OpenAI Gym 中级教程：多智能体系统

在强化学习中，多智能体系统涉及到多个智能体相互作用的情况。在本篇博客中，我们将介绍如何在 OpenAI Gym 中构建和训练多智能体系统，并使用 Multi-Agent Deep Deterministic Policy Gradients（MADDPG）算法进行协同训练。

1. 安装依赖

首先，确保你已经安装了 OpenAI Gym 和其他必要的依赖：

pip install gym
pip install numpy
pip install tensorflow
pip install matplotlib

2. 多智能体环境

我们将以一个简单的多智能体环境为例，该环境称为 MultiAgentEnv，其中包含两个智能体，它们分别控制两辆小车，目标是使两辆小车在一个二维平面上协同移动，避免相互碰撞。

import gym
from gym import spaces
import numpy as np

class MultiAgentEnv(gym.Env):
    def __init__(self):
        super(MultiAgentEnv, self).__init__()

        # 定义动作空间和观察空间
        self.action_space = spaces.Discrete(5)  # 5个离散动作
        self.observation_space = spaces.Box(low=0, high=1, shape=(4,), dtype=np.float32)  # 连续观察空间，包含4个状态

        # 初始化两辆小车的状态
        self.agent1_state = np.array([0.2, 0.5, 0.0, 0.0])
        self.agent2_state = np.array([0.8, 0.5, 0.0, 0.0])

    def reset(self):
        # 重置环境，将两辆小车放置在初始位置
        self.agent1_state = np.array([0.2, 0.5, 0.0, 0.0])
        self.agent2_state = np.array([0.8, 0.5, 0.0, 0.0])
        return np.concatenate([self.agent1_state, self.agent2_state])

    def step(self, actions):
        # 执行动作，更新两辆小车的状态并返回奖励和观察结果
        self.agent1_state[0] += 0.1 * (actions[0] - 2)
        self.agent1_state[1] += 0.1 * (actions[1] - 2)

        self.agent2_state[0] += 0.1 * (actions[2] - 2)
        self.agent2_state[1] += 0.1 * (actions[3] - 2)

        # 规定状态范围在 [0, 1] 之间
        self.agent1_state[:2] = np.clip(self.agent1_state[:2], 0, 1)
        self.agent2_state[:2] = np.clip(self.agent2_state[:2], 0, 1)

        # 计算奖励
        reward1 = -np.linalg.norm(self.agent1_state[:2] - self.agent2_state[:2])
        reward2 = -np.linalg.norm(self.agent2_state[:2] - self.agent1_state[:2])

        # 返回观察结果、奖励、是否终止和其他信息
        return np.concatenate([self.agent1_state, self.agent2_state]), [reward1, reward2], False, {
   }

3. MADDPG 算法

接下来，我们将实现 MADDPG 算法。为了简化，我们将只实现两个智能体的情况。

import tensorflow as tf
from tensorflow.keras import layers

class ActorCritic(tf.keras.Model):
    def __init__(self, num_actions):
        super(ActorCritic, self).__init__()

        # 定义Actor网络
        self.actor_fc1 = layers.Dense(64, activation='relu')
        self.actor_fc2 = layers.Dense(64, activation='relu')
        self.actor_output = layers.Dense(num_actions, activation='softmax')

        # 定义Critic网络
        self.critic_fc1 = layers.Dense(64, activation='relu')
        self.critic_fc2 = layers.Dense(64, activation='relu')
        self.critic_output = layers.Dense(1, activation='linear')

    def call(self, state):
        # Actor网络输出动作概率
        actor_x = self.actor_fc1(state)
        actor_x = self.actor_fc2(actor_x)
        action_probs = self.actor_output(actor_x)

        # Critic网络输出状态值
        critic_x = self.critic_fc1(state)
        critic_x = self.critic_fc2(critic_x)
        state_value = self.critic_output(critic_x)

        return action_probs, state_value

4. 训练多智能体系统

现在，我们将使用 MADDPG 算法来训练多智能体系统。

def train_maddpg(env, model1, model2, optimizer1, optimizer2, num_episodes=1000, gamma=0.99):
    for episode in range(num_episodes):
        state = env.reset()
        state = tf.convert_to_tensor(state, dtype=tf.float32)

        total_reward1 = 0
        total_reward2 = 0

        with tf.GradientTape() as tape1, tf.GradientTape() as tape2:
            for t in range(1000):  # 最多运行1000个时间步
                action_probs1, state_value1 = model1(state[None, :])
                action1 = tf.random.categorical(tf.math.log(action_probs1), 1)[0, 0]

                action_probs2, state_value2 = model2(state[None, :])
                action2 = tf.random.categorical(tf.math.log(action_probs2), 1)[0, 0]

                next_state, rewards, done, _ = env.step([action1.numpy(), action2.numpy()])
                next_state = tf.convert_to_tensor(next_state, dtype=tf.float32)

                total_reward1 += rewards[0]
                total_reward2 += rewards[1]

                action_probs1_next, _ = model1(next_state[None, :])
                action_probs2_next, _ = model2(next_state[None, :])

                # 计算Advantage和Target
                advantage1 = rewards[0] + gamma * tf.reduce_max(action_probs1_next) - state_value1
                advantage2 = rewards[1] + gamma * tf.reduce_max(action_probs2_next) - state_value2

                target1 = rewards[0] + gamma * tf.reduce_max(action_probs1_next)
                target2 = rewards[1] + gamma * tf.reduce_max(action_probs2_next)

                # 计算Actor和Critic的损失
                loss_actor1 = -tf.math.log(action_probs1[0, action1]) * advantage1
                loss_actor2 = -tf.math.log(action_probs2[0, action2]) * advantage2

                loss_critic1 = tf.square(target1 - state_value1)
                loss_critic2 = tf.square(target2 - state_value2)

                # 计算总损失
                total_loss1 = loss_actor1 + loss_critic1
                total_loss2 = loss_actor2 + loss_critic2

        # 更新参数
        gradients1 = tape1.gradient(total_loss1, model1.trainable_variables)
        optimizer1.apply_gradients(zip(gradients1, model1.trainable_variables))

        gradients2 = tape2.gradient(total_loss2, model2.trainable_variables)
        optimizer2.apply_gradients(zip(gradients2, model2.trainable_variables))

        if episode % 10 == 0:
            print(f"Episode: {episode}, Total Reward Agent 1: {total_reward1}, Total Reward Agent 2: {total_reward2}")

5. 主函数

最后，我们将定义一个主函数来运行我们的多智能体系统。

if __name__ == "__main__":
    # 创建多智能体环境和模型
    env = MultiAgentEnv()
    model1 = ActorCritic(num_actions=5)
    model2 = ActorCritic(num_actions=5)

    # 创建优化器
    optimizer1 = tf.optimizers.Adam(learning_rate=0.001)
    optimizer2 = tf.optimizers.Adam(learning_rate=0.001)

    # 训练多智能体系统
    train_maddpg(env, model1, model2, optimizer1, optimizer2, num_episodes=500)

通过这个示例，我们演示了如何在 OpenAI Gym 中构建一个简单的多智能体环境，并使用 MADDPG 算法对多智能体系统进行协同训练。这个示例可以作为入门多智能体强化学习的起点，同时展示了 TensorFlow 和 OpenAI Gym 在多智能体环境中的基本应用。希望这篇博客对你理解和应用多智能体系统有所帮助。