Не удалось достичь такой же хорошей производительности PPO с MAPPO.

У меня есть многопользовательская среда, основанная на боидов, использующая gymnasium. Я использовал PPO с политикой Mlp и сетью из 8 слоев по 512 нейронов в каждом слое. Хотя я достиг объединения, это было централизованное решение.

Результаты:
Объединение (Централизованный PPO)

Сеть:

policy_kwargs = dict(
    activation_fn=th.nn.Tanh,  # Использование функции активации ReLU
    net_arch=[dict(pi=[512, 512, 512, 512, 512, 512, 512, 512], 
                   vf=[512, 512, 512, 512, 512, 512, 512, 512])]  # Отдельные сети для политики (pi) и функции ценности (vf)
)

Код обучения:

model = PPO("MlpPolicy", env, kwargs=policy_kwargs tensorboard_log="./ppo_Agents_tensorboard/", verbose=1, device=device)
model.set_random_seed(SimulationVariables["ModelSeed"])

Моя цель, однако, состоит в том, чтобы создать децентрализованную версию Это по сути MAPPO (PPO с децентрализованными акторами, но с общим критиком, как утверждается, это хорошо для группового поведения, в отличие от IPPO — полностью децентрализованного). Вот почему я создал индивидуальную политику сети следующим образом с SB3:

class CustomMultiAgentPolicy(ActorCriticPolicy):  # Убедитесь, что наследуетесь от правильного родительского класса
    def __init__(self, observation_space, action_space, lr_schedule, **kwargs):
        super(CustomMultiAgentPolicy, self).__init__(observation_space, action_space, lr_schedule, **kwargs)

        self.obs_size = observation_space.shape[0] # только это в соответствии с вашей структурой наблюдения
        self.hidden_size = 128  # Вы можете изменить это в зависимости от ваших потребностей
        self.action_space = action_space

        self.actor = CustomActor(observation_space, action_space)
        self.critic = SharedCritic(observation_space)

    def forward(self, obs, **kwargs):
        action_mean = self.actor(obs)  # Это среднее нормального распределения

        # Создайте нормальное распределение со средним значением и log_std (преобразованным в std)
        # action_std = th.exp(self.log_std)  # Преобразуйте log_std в std
        action_std = th.clamp(th.exp(self.log_std), min=1e-3, max=1.0)  # Стабилизируйте std

        action_distribution = th.distributions.Normal(action_mean, action_std)

        # Выбор действий и получение логарифмических вероятностей
        actions = action_distribution.sample()
        log_probs = action_distribution.log_prob(actions).sum(dim=-1)  # Сумма по измерениям действия

        values = self.critic(obs)

        # Убедитесь, что действия соответствуют ожидаемой форме для среды (в вашем случае, [1, 6])
        return actions, values, log_probs

Независимый актор и общий критик:

class CustomActor(th.nn.Module):
    def __init__(self, observation_space, action_space):
        super(CustomActor, self).__init__()
        self.device = th.device("cuda" if th.cuda.is_available() else "cpu")

        # Создайте 8 слоев с 512 нейронами каждый
        self.layers = th.nn.ModuleList()
        input_size = observation_space.shape[0]

        for _ in range(8):
            layer = th.nn.Linear(input_size, 512).to(self.device)  # Переместите слой на устройство
            self.layers.append(layer)
            input_size = 512  # Обновите размер ввода для следующего слоя

        # Обновите голову действия в зависимости от типа пространства действий
        if isinstance(action_space, spaces.Box):  # Непрерывное пространство действий
            self.action_head = th.nn.Linear(512, action_space.shape[0]).to(self.device)
        elif isinstance(action_space, spaces.Discrete):  # Дискретное пространство действий
            self.action_head = th.nn.Linear(512, action_space.n).to(self.device)
        else:
            raise NotImplementedError("Тип пространства действий не поддерживается")

    def forward(self, x):
        # Преобразуйте ввод в тензор torch, если это массив numpy, и переместите его на правильное устройство
        if isinstance(x, np.ndarray):
            x = th.tensor(x, dtype=th.float32).to(self.device)  # Переместите на устройство здесь

        # Пропустите ввод через слои сети
        for layer in self.layers:
            # print(f"layer weight device: {layer.weight.device}")  # Печать внутри цикла
            x = F.relu(layer(x))  # Все слои должны уже находиться на self.device

        # Получите логиты действий из головы действия
        action_logits = self.action_head(x)  # action_head также должен находиться на том же устройстве

        return action_logits

class SharedCritic(th.nn.Module):
    def __init__(self, observation_space):
        super(SharedCritic, self).__init__()

        # Создайте 8 слоев с 512 нейронами каждый
        self.layers = th.nn.ModuleList()
        input_size = observation_space.shape[0]

        for _ in range(8):
            self.layers.append(th.nn.Linear(input_size, 512))  # Переместите на устройство в методе forward, если это необходимо
            input_size = 512  # Обновите размер ввода для следующего слоя

        self.value_head = th.nn.Linear(512, 1)

    def forward(self, x):
        # Преобразуйте ввод в тензор torch, если это массив numpy
        if isinstance(x, np.ndarray):
            x = th.tensor(x, dtype=th.float32)

        for layer in self.layers:
            x = F.relu(layer(x))
        value = self.value_head(x)
        return value

Мои результаты с этим, несмотря на то что я абсолютно ничего не изменил, кроме сети, составляют 180 к тому, что должно быть (пробовал с аналогичным количеством временных шагов и меньшим тоже).

Результаты (MAPPO) Они движутся в совершенно разные стороны, несмотря на то, что моя функция вознаграждения осталась неизменной.

Функция вознаграждения:

CohesionReward = 0
        AlignmentReward = 0
        total_reward = 0
        outofflock = False
        midpoint = (SimulationVariables["SafetyRadius"] + SimulationVariables["NeighborhoodRadius"]) / 2

        if len(neighbor_positions) > 0:
            for neighbor_position in neighbor_positions:
                distance = np.linalg.norm(agent.position - neighbor_position)

                if distance <= SimulationVariables["SafetyRadius"]:
                    CohesionReward -= 10
                elif SimulationVariables["SafetyRadius"] < distance <= midpoint:
                    CohesionReward += 20 - (20 / (midpoint - SimulationVariables["SafetyRadius"])) * (distance - SimulationVariables["SafetyRadius"])
                elif midpoint < distance <= SimulationVariables["NeighborhoodRadius"]:
                    CohesionReward += 10 - (10 / (SimulationVariables["NeighborhoodRadius"] - midpoint)) * (distance - midpoint)

                average_velocity = np.mean(neighbor_velocities, axis = 0)
                dot_product = np.dot(average_velocity, agent.velocity)
                norm_product = np.linalg.norm(average_velocity) * np.linalg.norm(agent.velocity)

                if norm_product == 0:
                    cos_angle = 1.0
                else:
                    cos_angle = dot_product / norm_product

                cos_angle = np.clip(cos_angle, -1.0, 1.0)
                orientation_diff = np.arccos(cos_angle)
                alignment = (orientation_diff / np.pi)
                AlignmentReward = -20 * alignment + 10  

        else:
            CohesionReward -= 20
            outofflock = True

        total_reward = CohesionReward + AlignmentReward
return total_reward, outofflock