PettingZoo框架

它是什么

简单来说：PettingZoo 就是多智能体版本的 Gym。
它由 Farama Foundation 维护（这个组织现在也维护着 Gymnasium，也就是原来的 OpenAI Gym）。如果说 Gymnasium 是单人游戏的标准，那么 PettingZoo 就是多人游戏的标准库。

为什么它是 Windows 用户的福音

• 纯 Python 编写：除了极少数依赖（如 Atari），它的核心库几乎全是 Python。这意味着：没有复杂的 C++ 编译过程，没有 CMake 报错，pip install 直接用。
• 标准化接口：它的用法和 Gymnasium 极其相似。如果你会写 env.step() 和 env.reset()，那你几乎不需要任何学习成本就能上手 PettingZoo。
• 生态丰富：它内置了大量环境，从简单的经典控制，到 Atari 游戏，再到复杂的粒子物理模拟。

它的两大API模式

ettingZoo 设计非常精妙，支持两种模式：

• AEC (Actor Environment Cycle)：回合制。
  就像下棋，你一步，我一步。适合：棋牌类、顺序决策任务。
• Parallel (并行)：同时行动。
  就像踢足球，大家同时跑。适合：绝大多数强化学习算法（PPO, MADDPG）。

安装和简单演示

安装

pip install pettingzoo[mpe] stable-baselines3 supersuit

演示骨架 (动作随机/无学习算法/跑完一局就结束)

from pettingzoo.mpe import simple_adversary_v3

# 创建环境：简单的对抗游戏（1个坏人，2个好人）
env = simple_adversary_v3.env(render_mode="human")
env.reset(seed=42)

# PettingZoo 的核心循环：Agent Iteration
for agent in env.agent_iter():
    observation, reward, termination, truncation, info = env.last()
    
    if termination or truncation:
        action = None
    else:
        # 这里替换成你的 AI 策略
        action = env.action_space(agent).sample() 

    env.step(action)
    
env.close()

完整演示

"""
PettingZoo 多智能体强化学习训练示例
环境：Simple Adversary（简单对抗游戏）
算法：PPO (Proximal Policy Optimization)
"""

import numpy as np
from pettingzoo.mpe import simple_adversary_v3
import supersuit as ss
from stable_baselines3 import PPO
from stable_baselines3.ppo import MlpPolicy
import os

# ============ 配置 ============
TRAIN_STEPS = 100_000      # 训练步数（可以调大到 500_000 效果更好）
MODEL_PATH = "adversary_model"  # 模型保存路径


def make_env():
    """创建并预处理环境"""
    env = simple_adversary_v3.parallel_env(max_cycles=25, continuous_actions=True)
    
    # SuperSuit 预处理：
    # 1. 将观察值填充到相同大小（不同智能体观察空间可能不同）
    env = ss.pad_observations_v0(env)
    # 2. 将动作空间填充到相同大小
    env = ss.pad_action_space_v0(env)
    # 3. 转换为 SB3 兼容的向量化环境
    env = ss.pettingzoo_env_to_vec_env_v1(env)
    # 4. 堆叠多个环境并行训练（加速）
    env = ss.concat_vec_envs_v1(env, num_vec_envs=8, num_cpus=1, base_class="stable_baselines3")
    
    return env


def train():
    """训练模型"""
    print("=" * 50)
    print("🚀 开始训练 Simple Adversary 环境")
    print("=" * 50)
    
    env = make_env()
    
    # 创建 PPO 模型
    model = PPO(
        MlpPolicy,
        env,
        verbose=1,
        learning_rate=1e-3,
        batch_size=256,
        n_steps=256,
        gamma=0.99,
        tensorboard_log="./logs/"
    )
    
    # 开始训练
    model.learn(total_timesteps=TRAIN_STEPS, progress_bar=True)
    
    # 保存模型
    model.save(MODEL_PATH)
    print(f"\n✅ 模型已保存到: {MODEL_PATH}.zip")
    
    env.close()
    return model


def test(model=None, num_episodes=3):
    """测试训练好的模型"""
    print("\n" + "=" * 50)
    print("🎮 测试训练好的 AI")
    print("=" * 50)
    
    # 加载模型
    if model is None:
        if os.path.exists(f"{MODEL_PATH}.zip"):
            model = PPO.load(MODEL_PATH)
            print(f"📦 已加载模型: {MODEL_PATH}.zip")
        else:
            print("❌ 找不到模型文件，请先训练！")
            return
    
    # 创建可视化环境
    env = simple_adversary_v3.parallel_env(render_mode="human", max_cycles=25, continuous_actions=True)
    env = ss.pad_observations_v0(env)
    env = ss.pad_action_space_v0(env)
    
    for episode in range(num_episodes):
        print(f"\n--- Episode {episode + 1}/{num_episodes} ---")
        observations, infos = env.reset()
        
        total_rewards = {agent: 0 for agent in env.agents}
        
        while env.agents:
            # 使用训练好的模型预测动作
            actions = {}
            for agent in env.agents:
                obs = observations[agent]
                action, _ = model.predict(obs, deterministic=True)
                actions[agent] = action
            
            observations, rewards, terminations, truncations, infos = env.step(actions)
            
            # 累计奖励
            for agent, reward in rewards.items():
                total_rewards[agent] += reward
        
        print(f"总奖励: {total_rewards}")
    
    env.close()
    print("\n🏁 测试完成！")


def test_random(num_episodes=2):
    """用随机策略测试（对比用）"""
    print("\n" + "=" * 50)
    print("🎲 随机策略测试（对比用）")
    print("=" * 50)
    
    env = simple_adversary_v3.parallel_env(render_mode="human", max_cycles=25, continuous_actions=True)
    
    for episode in range(num_episodes):
        print(f"\n--- Episode {episode + 1}/{num_episodes} ---")
        observations, infos = env.reset()
        
        total_rewards = {agent: 0 for agent in env.agents}
        
        while env.agents:
            # 随机动作
            actions = {agent: env.action_space(agent).sample() for agent in env.agents}
            observations, rewards, terminations, truncations, infos = env.step(actions)
            
            for agent, reward in rewards.items():
                total_rewards[agent] += reward
        
        print(f"总奖励: {total_rewards}")
    
    env.close()


if __name__ == "__main__":
    
    model = train()
    test(model)