3. 参数共享与不共享对比

任务目标

在CTDE（集中训练、分散执行）框架下，实现一个多智能体协作环境，分别训练参数共享与参数不共享两组智能体，对比两者在相同环境下的学习效率（收敛所需步数）与最终性能（平均回报）。

环境与依赖

• Python 3.8+，仅依赖NumPy（用于环境与训练循环）
• 不使用PyTorch/TensorFlow，用NumPy手动实现简易策略梯度，保证逻辑透明、可独立运行

核心实现

环境为2个智能体协作推动一个共享目标（类似简易协作搬运）：每个智能体观察自身与目标的相对位置，动作空间为离散的4个方向（上下左右）。只有两个智能体同时向目标方向施加力时，目标才会移动，否则原地不动。到达目标位置（距离<0.1）则回合结束，获得团队奖励+10，每步有-0.05的小惩罚以鼓励快速完成。

策略网络为单层线性softmax。参数共享模式下，两个智能体使用同一组策略参数；不共享模式下各自维护独立参数。训练使用REINFORCE算法，基线为回合内累计奖励的指数滑动平均。

import numpy as np

# ---------- 环境 ----------
class CooperativeEnv:
    def __init__(self):
        self.goal = np.array([0.8, 0.8])
        self.agents = None
        self.reset()

    def reset(self):
        self.agents = np.array([[0.1, 0.1], [0.1, 0.2]])  # 两个智能体初始位置
        self.steps = 0
        return self._obs()

    def _obs(self):
        # 每个智能体观察：自身位置 + 目标相对位置
        return np.concatenate([self.agents, np.tile(self.goal - self.agents[0], (2, 1))], axis=1).flatten()

    def step(self, actions):
        # actions: [a0, a1]，0=左 1=右 2=下 3=上
        delta = np.array([[-0.1, 0], [0.1, 0], [0, -0.1], [0, 0.1]])
        move0 = delta[actions[0]]
        move1 = delta[actions[1]]
        # 协作条件：两个智能体必须朝同一方向移动（同向才有效）
        if np.array_equal(move0, move1):
            self.agents += move0
        self.steps += 1
        dist = np.linalg.norm(self.agents[0] - self.goal)
        done = dist < 0.1 or self.steps > 100
        reward = 10.0 if dist < 0.1 else -0.05
        return self._obs(), reward, done

# ---------- 策略 ----------
class Policy:
    def __init__(self, obs_dim, n_actions, seed):
        self.rng = np.random.default_rng(seed)
        self.W = self.rng.normal(0, 0.1, (obs_dim, n_actions))

    def act_probs(self, obs):
        logits = obs @ self.W
        exp = np.exp(logits - logits.max())
        return exp / exp.sum()

    def act(self, obs):
        probs = self.act_probs(obs)
        return self.rng.choice(len(probs), p=probs)

# ---------- 训练 ----------
def train(share_params, episodes=800, lr=0.01, seed=42):
    env = CooperativeEnv()
    obs_dim = env._obs().shape[0]
    n_actions = 4
    rng = np.random.default_rng(seed)

    if share_params:
        policies = [Policy(obs_dim, n_actions, seed)]  # 共享同一策略
    else:
        policies = [Policy(obs_dim, n_actions, seed), Policy(obs_dim, n_actions, seed+1)]

    baseline = 0.0
    reward_history = []
    success_step_history = []

    for ep in range(episodes):
        obs = env.reset()
        log_probs = []
        rewards = []
        done = False
        while not done:
            obs = obs.reshape(1, -1)
            actions = []
            for i in range(2):
                p = policies[0] if share_params else policies[i]
                probs = p.act_probs(obs[0])
                a = rng.choice(n_actions, p=probs)
                actions.append(a)
                log_probs.append(np.log(probs[a] + 1e-10))
            obs, r, done = env.step(actions)
            rewards.append(r)

        G = sum(rewards)
        baseline = 0.9 * baseline + 0.1 * G if ep > 0 else G
        advantage = G - baseline

        # 策略梯度更新
        for lp in log_probs:
            grad = advantage * lp
            if share_params:
                policies[0].W += lr * grad * np.outer(obs[0], np.eye(n_actions)[actions[0]])  # 简化：实际需按动作求导
            # 注：上面为示意，实际梯度需按softmax导数计算，见下方修正

        reward_history.append(G)
        if G > 0:
            success_step_history.append(env.steps)

    return reward_history, success_step_history

# ---------- 修正后的梯度更新（替换上面train中的更新部分） ----------
def update_policy(policy, obs_list, action_list, advantage, lr):
    # obs_list: 每个时间步的观测 (T, obs_dim)
    # action_list: 每个时间步的动作 (T,)
    T = len(action_list)
    for t in range(T):
        probs = policy.act_probs(obs_list[t])
        grad = np.zeros_like(policy.W)
        for a in range(policy.W.shape[1]):
            grad[:, a] = -probs[a] * obs_list[t]
        grad[:, action_list[t]] += obs_list[t]
        policy.W += lr * advantage * grad

# 重新整理训练循环（修正版）
def train_corrected(share_params, episodes=800, lr=0.01, seed=42):
    env = CooperativeEnv()
    obs_dim = env._obs().shape[0]
    n_actions = 4
    rng = np.random.default_rng(seed)

    if share_params:
        policies = [Policy(obs_dim, n_actions, seed)]
    else:
        policies = [Policy(obs_dim, n_actions, seed), Policy(obs_dim, n_actions, seed+1)]

    baseline = 0.0
    reward_history = []
    success_step_history = []

    for ep in range(episodes):
        obs = env.reset()
        obs_traj = []
        act_traj = []
        rew_traj = []
        done = False
        while not done:
            obs_traj.append(obs.copy())
            actions = []
            for i in range(2):
                p = policies[0] if share_params else policies[i]
                probs = p.act_probs(obs)
                a = rng.choice(n_actions, p=probs)
                actions.append(a)
            act_traj.append(actions)
            obs, r, done = env.step(actions)
            rew_traj.append(r)

        G = sum(rew_traj)
        baseline = 0.9 * baseline + 0.1 * G if ep > 0 else G
        advantage = G - baseline

        # 分别更新每个智能体的策略
        for i in range(2):
            p = policies[0] if share_params else policies[i]
            acts_i = [a[i] for a in act_traj]
            update_policy(p, obs_traj, acts_i, advantage, lr)

        reward_history.append(G)
        if G > 0:
            success_step_history.append(env.steps)

    return reward_history, success_step_history

# ---------- 运行对比 ----------
if __name__ == "__main__":
    print("训练参数共享...")
    shared_rewards, shared_steps = train_corrected(True, episodes=500)
    print("训练参数不共享...")
    indep_rewards, indep_steps = train_corrected(False, episodes=500)

    # 输出对比指标
    def smooth(arr, w=50):
        if len(arr) < w: return np.mean(arr)
        return np.convolve(arr, np.ones(w)/w, mode='valid')

    print("\n===== 对比结果 =====")
    print(f"参数共享 - 前100回合平均回报: {np.mean(shared_rewards[:100]):.2f}")
    print(f"参数共享 - 后100回合平均回报: {np.mean(shared_rewards[-100:]):.2f}")
    print(f"参数共享 - 首次成功回合: {next((i for i,r in enumerate(shared_rewards) if r>0), '无')}")
    print(f"参数不共享 - 前100回合平均回报: {np.mean(indep_rewards[:100]):.2f}")
    print(f"参数不共享 - 后100回合平均回报: {np.mean(indep_rewards[-100:]):.2f}")
    print(f"参数不共享 - 首次成功回合: {next((i for i,r in enumerate(indep_rewards) if r>0), '无')}")

运行与验证

1. 将上述代码保存为 compare_sharing.py，直接运行 python compare_sharing.py
2. 观察输出中的对比指标：

判定标准：若参数共享组的首次成功回合编号明显小于不共享组（例如共享组在100回合内首次成功，不共享组在200回合后），且前100回合平均回报更高，则说明参数共享在该协作任务中提升了学习效率；若最终两组都能稳定获得正回报，则说明参数共享未损害最终性能上限。

关键说明

• 本案例使用2智能体协作环境，参数共享意味着两个智能体共用同一套策略网络参数，梯度更新时叠加两个智能体的经验；不共享则各自独立更新。
• 协作条件（同向移动才有效）确保任务需要智能体间协调，否则无法获得正奖励，从而能清晰区分两种模式的学习差异。
• 由于使用随机策略梯度，单次运行有波动。若差异不明显，可增大episodes至1000或调整随机种子（seed参数）后重跑。