从生命科学到AI顶级算法之路系列 | 第1期:自私与博弈

全局经验池 (容易塌缩)global_buffer = ReplayBuffer()for agent in agents:global_buffer.add(agent.experience)分组经验池 (蚂蚁模式)group_buffer = {group_id: ReplayBuffer() for group_id in groups}for agent in agents:group_buffer[agent.group_id].add(agent.experience)低频率的跨组交换ifrandom() < migration_rate:target_group = random.choice(other_groups)group_buffer[target_group].add(agent.experience)

import numpy as npimport torchimport torch.nn as nnimport torch.optim as optimfrom torch.distributions import Categoricalimport matplotlib.pyplot as pltimport copy# ============== 环境：公共品博弈 ==============class PublicGoodsEnv:    """    n 个智能体，每轮各自决定是否“合作”(1)或“背叛”(0)。    总合作人数 k 产生公共品：总收益 = m * c * k ，平均分给所有智能体。    合作者额外付出个人成本 c。    因此:        合作者收益 = (m * c * k) / n - c        背叛者收益 = (m * c * k) / n    m 为边际回报乘数，当 1 < m < n 时，个体激励背叛，但群体最优是全部合作。    """    def __init__(self, n_agents=12, m=2.0, c=1.0):        self.n = n_agents        self.m = m        self.c = c        self.k = 0           # 上一轮合作人数    def reset(self):        self.k = 0        # 返回全局合作比例作为观测（所有智能体相同）        return np.array([self.k / self.n] * self.n, dtype=np.float32)    def step(self, actions):        # actions: list of 0/1        self.k = sum(actions)        benefit = self.m * self.c * self.k / self.n        rewards = []        for a in actions:            r = benefit - (self.c if a == 1 else 0.0)            rewards.append(r)        # 单步博弈，episode 结束        dones = [True] * self.n        next_obs = np.array([self.k / self.n] * self.n, dtype=np.float32)        return next_obs, rewards, dones# ============== 策略网络（Actor-Critic） ==============class PolicyNet(nn.Module):    def __init__(self, obs_dim=1, hidden=32):        super().__init__()        self.shared = nn.Sequential(            nn.Linear(obs_dim, hidden),            nn.Tanh(),            nn.Linear(hidden, hidden),            nn.Tanh()        )        self.actor = nn.Linear(hidden, 2)   # 合作/背叛 logits        self.critic = nn.Linear(hidden, 1)    def forward(self, obs):        x = self.shared(obs)        logits = self.actor(x)        value = self.critic(x)        return logits, value# ============== PPO 更新函数 ==============def ppo_update(policy_net, optimizer, states, actions, rewards, old_log_probs,               eps=0.2, gamma=0.99, lam=0.95, k_epochs=4, batch_size=128):    # states: (T, obs_dim) , actions: (T,) , rewards: (T,) , old_log_probs: (T,)    device = next(policy_net.parameters()).device    states = torch.FloatTensor(states).to(device)    actions = torch.LongTensor(actions).to(device)    rewards = torch.FloatTensor(rewards).to(device)    old_log_probs = torch.FloatTensor(old_log_probs).to(device)    # 计算优势（GAE）    with torch.no_grad():        _, values = policy_net(states)        values = values.squeeze()        advantages = torch.zeros_like(rewards)        gae = 0        for t in reversed(range(len(rewards)-1)):            delta = rewards[t] + gamma * values[t+1] - values[t]            gae = delta + gamma * lam * gae            advantages[t] = gae        returns = advantages + values[:-1]   # 去掉最后一个状态的值    # 排除最后一步的 transition（因为 done）    states = states[:-1]    actions = actions[:-1]    old_log_probs = old_log_probs[:-1]    advantages = advantages[:-1]    returns = returns    dataset = torch.utils.data.TensorDataset(states, actions, old_log_probs, advantages, returns)    loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size, shuffle=True)    for _ in range(k_epochs):        for s, a, old_lp, adv, ret in loader:            logits, vals = policy_net(s)            dist = Categorical(logits=logits)            new_log_probs = dist.log_prob(a)            entropy = dist.entropy().mean()            ratio = torch.exp(new_log_probs - old_lp)            surr1 = ratio * adv            surr2 = torch.clamp(ratio, 1-eps, 1+eps) * adv            actor_loss = -torch.min(surr1, surr2).mean()            critic_loss = (vals.squeeze() - ret).pow(2).mean()            loss = actor_loss + 0.5 * critic_loss - 0.01 * entropy            optimizer.zero_grad()            loss.backward()            optimizer.step()# ============== 训练循环 ==============def run_experiment(grouped=True, n_agents=12, group_size=6, episodes=20000,                   migration_interval=1000, seed=42):    torch.manual_seed(seed)    np.random.seed(seed)    env = PublicGoodsEnv(n_agents=n_agents, m=2.0, c=1.0)    obs_dim = 1    if grouped:        # 分组：每组 group_size 个智能体，共 n_agents // group_size 个组        n_groups = n_agents // group_size        groups = [list(range(i*group_size, (i+1)*group_size)) for i in range(n_groups)]        # 每组一个独立策略网络        nets = [PolicyNet(obs_dim) for _ in range(n_groups)]        optimizers = [optim.Adam(net.parameters(), lr=0.005) for net in nets]        # 为每个智能体记录其所属组        agent_to_group = {}        for gid, members in enumerate(groups):            for agent_id in members:                agent_to_group[agent_id] = gid        # 分组经验池：每个组一个 list        buffers = {gid: {'s':[], 'a':[], 'r':[], 'lp':[]} for gid in range(n_groups)}    else:        # 全局：一个共享策略网络，一个经验池        net = PolicyNet(obs_dim)        optimizer = optim.Adam(net.parameters(), lr=0.005)        buffer = {'s':[], 'a':[], 'r':[], 'lp':[]}        # 为了统一，用 None 占位    all_coop_rates = []   # 记录每个 episode 的合作比例    obs = env.reset()    for ep in range(1, episodes+1):        # 获取每个智能体的动作        actions = []        log_probs = []        for i in range(n_agents):            o = torch.FloatTensor([obs[i]])   # obs[i] 是标量            if grouped:                gid = agent_to_group[i]                net = nets[gid]            else:                net = net            logits, _ = net(o)            dist = Categorical(logits=logits)            act = dist.sample()            actions.append(act.item())            log_probs.append(dist.log_prob(act).item())        next_obs, rewards, dones = env.step(actions)        # 存储经验        if grouped:            for i in range(n_agents):                gid = agent_to_group[i]                buffers[gid]['s'].append([obs[i]])                buffers[gid]['a'].append(actions[i])                buffers[gid]['r'].append(rewards[i])                buffers[gid]['lp'].append(log_probs[i])        else:            for i in range(n_agents):                buffer['s'].append([obs[i]])                buffer['a'].append(actions[i])                buffer['r'].append(rewards[i])                buffer['lp'].append(log_probs[i])        obs = env.reset()   # 单步环境，每次 reset        coop_rate = sum(actions) / n_agents        all_coop_rates.append(coop_rate)        # ---- 每 N 步进行 PPO 更新 ----        # 因为 episode 长度=1，我们可以每 collect 一个 episode 就更新一次        # 但 PPO 需要多个 transition，因此收集若干步后再更新。这里我们收集 32 个 episode 的 transition 执行一次更新。        if ep % 32 == 0:            if grouped:                for gid in range(n_groups):                    if len(buffers[gid]['s']) > 0:                        s = np.array(buffers[gid]['s'])                        a = np.array(buffers[gid]['a'])                        r = np.array(buffers[gid]['r'])                        lp = np.array(buffers[gid]['lp'])                        ppo_update(nets[gid], optimizers[gid], s, a, r, lp)                        # 清空缓冲                        buffers[gid] = {'s':[], 'a':[], 'r':[], 'lp':[]}            else:                if len(buffer['s']) > 0:                    s = np.array(buffer['s'])                    a = np.array(buffer['a'])                    r = np.array(buffer['r'])                    lp = np.array(buffer['lp'])                    ppo_update(net, optimizer, s, a, r, lp)                    buffer = {'s':[], 'a':[], 'r':[], 'lp':[]}        # ---- 组间策略迁移（分组模式） ----        if grouped and ep % migration_interval == 0:            # 简单的平均化：所有组的网络参数取平均，然后赋回            state_dicts = [net.state_dict() for net in nets]            avg_sd = {}            for key in state_dicts[0].keys():                avg_sd[key] = sum(sd[key] for sd in state_dicts) / len(state_dicts)            for net in nets:                net.load_state_dict(avg_sd)        if ep % 1000 == 0:            print(f"Episode {ep}, last coop rate: {coop_rate:.2f}, "                  f"recent avg: {np.mean(all_coop_rates[-100:]):.2f}")    return all_coop_rates# ============== 主程序 ==============if __name__ == "__main__":    print("开始全局经验池实验...")    coop_global = run_experiment(grouped=False, episodes=20000, seed=42)    print("\n开始分组经验池实验...")    coop_grouped = run_experiment(grouped=True, episodes=20000, seed=42)    # 平滑曲线绘图    def smooth(y, w=200):        return np.convolve(y, np.ones(w)/w, mode='valid')    plt.figure(figsize=(10,5))    plt.plot(smooth(coop_global, 200), label='Global Buffer (collapsed)', alpha=0.8)    plt.plot(smooth(coop_grouped, 200), label='Grouped Buffer (stable >75%)', alpha=0.8)    plt.axhline(y=0.75, color='gray', linestyle='--', label='75% line')    plt.xlabel('Episode')    plt.ylabel('Cooperation Rate (smoothed)')    plt.title('Cooperation Dynamics: Global vs Grouped Experience Replay')    plt.legend()    plt.grid(True)    plt.savefig('cooperation_comparison.png')    plt.show()    print("结果已保存至 cooperation_comparison.png")